首页 > 其他分享 >深入学习RabbitMQ五种模式(三)

深入学习RabbitMQ五种模式(三)

时间：2023-04-29 09:56:46浏览次数：32  
标签：String 队列 RabbitMQ 交换机 深入 key 五种 com channel

1.路由模式(精确匹配)

路由模式（Routing）的特点：

  • 该模式的交换机为direct，意思为定向发送，精准匹配。
  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
  • 消息的发送方在向Exchange发送消息时，也必须指定消息的 RoutingKey。
  • Exchange不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey进行判断，只有队列的Routingkey与消息的Routing key完全一致，才会接收到消息。

生产者将消息发送到direct交换器，同时生产者在发送消息的时候会指定一个路由key，而在绑定队列和交换器的时候又会指定一个路由key，那么消息只会发送到相应routing key相同的队列，然后由监听该队列的消费者进行消费消息。模型如下图所示：

  • 创建生产者
package com.olive;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * 生产者（路由模式）
 */
public class RoutorProducer {

    // 交换机名称
    private static final String EXCHANGE_NAME = "routing_exchange";

    public static void main(String[] args) throws Exception {
        // 1、创建连接
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道（频道）
        Channel channel = connection.createChannel();
        // 3、发送消息，连续发3条
        for (int i = 0; i < 3; i++) {
            String routingKey = "";
            //发送消息的时候根据相关逻辑指定相应的routing key。
            switch (i) {
                case 0:  //假设i=0，为error消息
                    routingKey = "error";
                    break;
                case 1: //假设i=1，为info消息
                    routingKey = "info";
                    break;
                case 2: //假设i=2，为warning消息
                    routingKey = "warning";
                    break;
            }
            // 要发送的消息
            String message = "Hello World Message!!!~~~" + routingKey;
            // 消息发送 channel.basicPublish(交换机名称,路由key,消息其它属性,消息内容)
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("utf-8"));
            System.out.println("生产者发送的消息：" + message);
        }
        //释放资源
        channel.close();
        connection.close();
    }
}
  • 创建消费者

消费者1

package com.olive;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 消费者1（路由模式）
 */
public class RoutorCunsumer1 {

    // 队列名称
    private static final String QUEUE_NAME1 = "routing_queue1";
    // 交换机名称
    private static final String EXCHANGE_NAME = "routing_exchange";

    public static void main(String[] args) throws Exception {
        // 1、获取连接对象
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道（频道）
        Channel channel = connection.createChannel();
        // 3、声明交换机(有则不创建,无则创建) channel.exchangeDeclare(交换机名字,交换机类型,是否持久化)
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
        // 4、声明队列Queue。channel.queueDeclare(队列名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
        channel.queueDeclare(QUEUE_NAME1, true, false, false, null);

        // 5、根据指定的routingKey绑定队列和交换机 channel.queueBind(队列名, 交换机名, 路由key)
        channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, "error");

        // 6、监听队列，接收消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //获取路由的key
                String routingKey = envelope.getRoutingKey();
                //获取交换机信息
                String exchange = envelope.getExchange();
                //获取消息信息
                String message = new String(body, "utf-8");
                System.out.println("路由Key:" + routingKey + ", 交换机名称:" + exchange + ", 消费者获取消息: " + message);
            }
        };
        channel.basicConsume(QUEUE_NAME1, true, defaultConsumer);
        //注意，消费者这里不建议关闭资源，让程序一直处于读取消息的状态
    }
}

消费者2

package com.olive;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 消费者2（路由模式）
 */
public class RoutorCunsumer2 {

    // 队列名称
    private static final String QUEUE_NAME2 = "routing_queue2";
    // 交换机名称
    private static final String EXCHANGE_NAME = "routing_exchange";

    public static void main(String[] args) throws Exception {
        // 1、获取连接对象
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道（频道）
        Channel channel = connection.createChannel();
        // 3、声明交换机(有则不创建,无则创建) channel.exchangeDeclare(交换机名字,交换机类型,是否持久化)
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
        // 4、声明队列Queue。channel.queueDeclare(队列名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
        channel.queueDeclare(QUEUE_NAME2, true, false, false, null);

        // 5、根据指定的routingKey绑定队列和交换机 channel.queueBind(队列名, 交换机名, 路由key)
        channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, "error");
        channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, "info");
        channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, "warning");

        // 6、监听队列，接收消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //获取路由的key
                String routingKey = envelope.getRoutingKey();
                //获取交换机信息
                String exchange = envelope.getExchange();
                //获取消息信息
                String message = new String(body, "utf-8");
                System.out.println("路由Key:" + routingKey + ", 交换机名称:" + exchange + ", 消费者获取消息: " + message);
            }
        };
        channel.basicConsume(QUEUE_NAME2, true, defaultConsumer);
    }
}
  • 验证

分别先启动所有消费者，再启动生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；达到按照需要接收的效果。

消费者1绑定的交换机和队列的路由Key为error，所以只要生产者发送消息时带有error的routingKey它都能够获取到消息。

消费者2绑定的交换机和队列的路由Key为error、info、warning，所以只要生产者发送消息时带有这3种的routingKey它都能够获取到消息。

从RabbitMQ管理后台也能看到对应的交换机，以及队列绑定情况

  • 总结
  1. Routing模式需要将交换机设置为Direct类型。
  2. Routing模式要求队列在绑定交换机时要指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列。

2.Topic模式(模糊匹配)

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。但是Topic类型的Exchange可以让队列在绑定Routing key的时候使用通配符进行匹配，也就是模糊匹配，这样与之前的模式比起来，它更加的灵活！

Topic主题模式的Routingkey一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： log.insert ，它的通配符规则如下：

*：匹配不多不少恰好1个词

#：匹配0或多个单词

简单举例：

log.*：只能匹配log.error，log.info等
log.#：能够匹配log.insert，log.insert.abc，log.news.update.abc等

图解：

  1. 红色Queue：绑定的是usa.# ，因此凡是以usa.开头的routing key都会被匹配到

  2. 黄色Queue：绑定的是#.news ，因此凡是以.news结尾的 routing key都会被匹配

  • 创建生产者
package com.olive;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * 生产者（Topic主题模式）
 */
public class TopicProducer {

    // 交换机名称
    private static final String EXCHANGE_NAME = "topic_exchange";

    public static void main(String[] args) throws Exception {
        // 1、创建连接
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道（频道）
        Channel channel = connection.createChannel();
        // 3、发送消息
        for (int i = 0; i < 4; i++) {
            String routingKey = "";
            //发送消息的时候根据相关逻辑指定相应的routing key。
            switch (i) {
                case 0:  //假设i=0，为select消息
                    routingKey = "log.select";
                    break;
                case 1: //假设i=1，为info消息
                    routingKey = "log.delete";
                    break;
                case 2: //假设i=2，为log.news.add消息
                    routingKey = "log.news.add";
                    break;
                case 3: //假设i=3，为log.news.update消息
                    routingKey = "log.news.update";
                    break;
            }
            // 要发送的消息
            String message = "Hello Message!!!~~~" + routingKey;
            // 消息发送 channel.basicPublish(交换机名称,路由key,消息其它属性,消息内容)
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("utf-8"));
            System.out.println("生产者发送的消息：" + message);
        }
        // 关闭资源
        channel.close();
        connection.close();
    }
}
  • 创建消费者

消费者1

接收所有与log.*相匹配的路由key队列中的消息

package com.olive;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 消费者（Topic模式）
 */
public class TopicConsumer1 {

    // 队列名称
    private static final String QUEUE_NAME1 = "topic_queue1";
    // 交换机名称
    private static final String EXCHANGE_NAME = "topic_exchange";

    public static void main(String[] args) throws Exception {
        // 1、获取连接对象
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道（频道）
        Channel channel = connection.createChannel();
        // 3、声明交换机(有则不创建,无则创建) channel.exchangeDeclare(交换机名字,交换机类型,是否持久化)
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, true);
        // 4、声明队列Queue channel.queueDeclare(队列名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
        channel.queueDeclare(QUEUE_NAME1, true, false, false, null);

        // 5、根据指定的routingKey绑定队列和交换机,设置路由key channel.queueBind(队列名, 交换机名, 路由key)
        channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, "log.*");

        // 6、监听队列，接收消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //获取路由的key
                String routingKey = envelope.getRoutingKey();
                //获取交换机信息
                String exchange = envelope.getExchange();
                //获取消息信息
                String message = new String(body, "utf-8");
                System.out.println("路由Key:" + routingKey + ", 交换机名称:" + exchange + ", 消费者获取消息: " + message);
            }
        };
        channel.basicConsume(QUEUE_NAME1, true, defaultConsumer);

        //注意，消费者这里不建议关闭资源，让程序一直处于读取消息的状态
    }
}

消费者2

接收所有与`log.``#相匹配的路由key队列中的消息

package com.olive;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 消费者（Topic模式）
 */
public class TopicConsumer2 {

    // 队列名称
    private static final String QUEUE_NAME2 = "topic_queue2";
    // 交换机名称
    private static final String EXCHANGE_NAME = "topic_exchange";

    public static void main(String[] args) throws Exception {
        // 1、获取连接对象
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道（频道）
        Channel channel = connection.createChannel();
        // 3、声明交换机(有则不创建,无则创建) channel.exchangeDeclare(交换机名字,交换机类型,是否持久化)
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, true);
        // 4、声明队列Queue。channel.queueDeclare(队列名称，是否持久化，是否独占本连接,是否自动删除,附加参数)
        channel.queueDeclare(QUEUE_NAME2, true, false, false, null);

        // 5、根据指定的routingKey绑定队列和交换机 channel.queueBind(队列名, 交换机名, 路由key)
        channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, "log.#");

        // 6、监听队列，接收消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //获取路由的key
                String routingKey = envelope.getRoutingKey();
                //获取交换机信息
                String exchange = envelope.getExchange();
                //获取消息信息
                String message = new String(body, "utf-8");
                System.out.println("路由Key:" + routingKey + ", 交换机名称:" + exchange + ", 消费者获取消息: " + message);
            }
        };
        channel.basicConsume(QUEUE_NAME2, true, defaultConsumer);
    }
}

分别先启动所有消费者，再使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；达到按照需要接收的效果。

消费者1的路由key匹配规则为log.*，所有该路由规则的绑定的队列应该只有2条信息。

消费者2的路由key匹配规则为log.#，它能够匹配以log.开头的所有路由key，所有该路由规则的绑定的队列应该只有4条信息。

从RabbitMQ管理后台也能看到对应的交换机，以及队列绑定情况

  • 总结

Topic主题模式需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列。

Topic主题模式可以实现Publish/Subscribe发布与订阅模式 和Routing路由模式的功能；只是Topic在配置routing key 的时候可以使用通配符，所以更加灵活。

来源
cnblogs.com/tanghaorong/p/14992330.html#_label0

标签：String,队列,RabbitMQ,交换机,深入,key,五种,com,channel
From： https://www.cnblogs.com/happyhuangjinjin/p/17363600.html

相关文章

  • 深入学习RabbitMQ五种模式(一)
    1.安装erlang下载otp_win64_25.3.exehttps://www.erlang.org/downloadserlang安装完成，需要配置erlang环境变量ERLANG_HOME=E:\software\ErlangOTPPATH=%PATH%;%ERLANG_HOME%\bin;2.安装RabbitMQ下载rabbitmq-server-3.11.13.exehttps://www.rabbitmq.com/download.htm......
  • SpringCloud Stream集成RabbitMQ
    1.概述SpringCloudStream框架抽象出了三个最基础的概念来对各种消息中间件提供统一调用：DestinationBinders:负责集成外部消息系统的组件。DestinationBinding:由Binders创建的，负责沟通外部消息系统、消息发送者和消息消费者的桥梁。Message:消息发送者与消息消费......
  • rabbitMQ--类型
    1.五种消息模型1.1基本消息模型 1.2work消息模型 1.3订阅模型1.3.1Fanout，也称为广播。流程说明流程图： 在广播模式下，消息发送流程是这样的：1）可以有多个消费者2）每个消费者有自己的queue（队列）3）每个队列都要绑定到Exchange（交换机）4）生产者发送的消息，只能......
  • 深入探讨源码--ArrayList
    持续推送技术干货目录深入探讨源码之ArrayListArrayList类图ArrayList的数据结构ArrayList的关键属性ArrayList构造方法ArrayList常用方法add方法ArrayList中的fast-fail机制add(i,o)方法set(i,o)方法get(i)方法remove(index)方法remove(Object)方法clear方法indexOf(o)方法深......
  • RabbitMQ _ How to Close a Channel
    https://low-orbit.net/rabbitmq-how-to-close-a-channel RabbitMQHowtoCloseaChannelIfyouhavefoundyourwaytothispageyouareprobablywonderinghowtocloseachannelinRabbitMQ.Channelsshouldbeclosedwhentheyarenolongerinuse.There......
  • 深入jar包：从jar包中读取资源文件
    我们常常在代码中读取一些资源文件(比如图片，音乐，文本等等)。在单独运行的时候这些简单的处理当然不会有问题。但是，如果我们把代码打成一个jar包以后，即使将资源文件一并打包，这些东西也找不出来了。看看下面的代码：Java代码 //源代码1：packageedu.hxraid;import......
  • Linux安装RabbitMQ
    前言：还是和以前一样，linux安装软件的目录都是data目录 1.这次稍微不一样，不过还是进入data目录，创建RabbitMq目录并进入该目录cd/datamkdirrabbitMqcdrabbitMq 2.上传"erlang-21.1-1.el7.x86_64.rpm"文件和 "rabbitmq-server-3.7.7-1.el7.noarch.rpm"文件到当前......
  • 字节前端--深入JS
    首先先介绍JS的基本概念：比如是单线程，动态，弱类型等等。除了这些东西之外还有：下面的一些基础概念：JavaScript是一种脚本语言，通常在网页上运行。JavaScript不需要编译，因为它是一种解释性语言。在网页上添加JavaScript的方式有多种，包括内联脚本、嵌入式脚本和外部脚本。......
  • 关于聚合根，领域事件的那点事---深入浅出理解DDD
    作者：京东物流赵勇萍前言最近有空会跟同事讨论DDD架构的实践落地的情况，但真实情况是，实际中对于领域驱动设计中的实体，值对象，聚合根，领域事件这些战术类的实践落地，每个人理解依然因人而异，大概率是因为这些概念还是有一些抽象，同时有有别于传统的MVC架构开发。在此，通过小demo的方式......
  • 用友iuap 让企业数智化能力深入、让业务价值浅出
     本文转自科技正能量作者郑凯IDC近期的一份调研中提出：数字化转型在2023年开始进入到一个新的阶段：从数字化转型时代，进入到数字化业务时代。在分水岭的两端，一个变化是：数字化转型时代是业务的数字化，而数字化业务时代则是数字的业务化。这个潜移默化的改变背后，其实有一重很深的含义。......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99