RabbitMQ系列【4】六大工作模式

曾经终败给现在 2024-04-06 14:07 33阅读 0赞

> **有道无术，术尚可求，有术无道，止于术。**

#### 文章目录 ####

*   *  前言
     *  简单模式
     *   *  1. 添加依赖
         *  2. 发送消息
         *  3. 接收消息
         *  4. 测试
     *  工作队列模式
     *   *  案例演示
     *  发布/订阅模式
     *  路由模式
     *  主题模式
     *  RPC 模式

### 前言 ###

`工作模式`指的是消息发送及接受的策略。

`RabbitMQ` 支持六种模式，官网图示：  
![在这里插入图片描述][7f602f427192406aad665541ca1e7dc3.png]

### 简单模式 ###

接下里我们使用`RabbitMQ` JAVA 客户端，实现一个简单的发送/接收消息案例。  
![在这里插入图片描述][2e315c569a8c483aa2aa256854a89f37.png]  
[RabbitMQ Java Client GitHub 地址][RabbitMQ Java Client GitHub]

#### 1. 添加依赖 ####

客户端5.X 版本，需要JDK 1.8+、RabbitMQ 服务端3.X 支持。

<dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.16.0</version>
    </dependency>

#### 2. 发送消息 ####

首先需要创建连接工厂，配置服务端地址信息：

// 1. 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");

客户端和服务端进行通信时，会使用`Connection`建立 TCP 连接。然后每一次通信时，会在`Connection`内部建立的逻辑连接`Channel`，这样可以大量减少TCP 连接开销。  
![在这里插入图片描述][d64ab0145dbe47d19e137fcd5c6955c9.png]  
所以接下来需要创建`Connection`和`Channel`：

// 2. 新建连接
            Connection connection = factory.newConnection();
            // 3. 新建通道
            Channel channel = connection.createChannel();

接着声明队列：

channel.queueDeclare("hello", false, false, false, null);

发送消息：

channel.basicPublish("", "hello", null, "hello world".getBytes());

完整代码如下：

public class Producer {
        
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 1. 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 2. 新建连接
            Connection connection = factory.newConnection();
            // 3. 新建通道
            Channel channel = connection.createChannel();
            /*
             * 4. 声明队列，参数：
             *  1.队列名称
             *  2.队列里面的消息是否持久化 默认消息存储在内存中
             *  3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
             *  4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
             *  5.其他参数
             */
            channel.queueDeclare("hello", false, false, false, null);
            /*
             * 5. 发送消息，参数：
             *  1.发送到那个交换机，（为空表示使用默认交换机）
             *  2.路由的 key ，默认交换隐式绑定到每个队列，路由KEY等于队列名称。无法显式绑定到默认交换或从默认交换解除绑定。它也不能被删除。
             *  3.其他的参数信息
             *  4.发送消息的消息体
             */
            channel.basicPublish("", "hello", null, "hello world".getBytes());
            System.out.println("消息发送完毕");
            // 6. 关闭连接
            channel.close();
            connection.close();
        }
    }

#### 3. 接收消息 ####

代码如下：

public class Consumer {
        
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 1. 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 2. 创建连接及通道
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
            // 3. 绑定队列
            channel.queueDeclare("hello", false, false, false, null);
            // 4. 处理消息的回调对象
            DeliverCallback deliverCallback = (consumerTag, delivery) ->
            {
        
                String message = new String(delivery.getBody());
                System.out.println(message);
            };
            // 4. 取消消息处理的回调对象
            CancelCallback cancelCallback = s -> {
        
                System.out.println("消息消费被中断");
            };
            // 5. 接收消息，把消息传递到回调对象处理，参数：
            /*  1.消费哪个队列
                2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
                3.消费者未成功消费的回调
             */
            channel.basicConsume("hello", true, deliverCallback, cancelCallback);
        }
    }

#### 4. 测试 ####

运行`Consumer`，再运行`Producer`，成功接收到消息：  
![在这里插入图片描述][27a152371a774116a23d998b97ccbc11.png]

### 工作队列模式 ###

`Work Queues`是官方中提出的第二种工作模式，一个生产者发送消息，有多个消费者来监听任务，但是只有一个消费者能收到消息：  
![在这里插入图片描述][0ac513f7b2e14cad9ab5638653ca4619.png]

`工作队列/任务队列`的主要思想是把任务封装为消息并将其发送到队列。在后台运行的工作进程将接收到任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。从而能实现异步多线程任务。

这个概念在web应用程序中特别有用，因为在短的HTTP请求窗口中不可能处理复杂的任务。

#### 案例演示 ####

复制一份`Consumer`类，分别启动：  
![在这里插入图片描述][195ad863d09d4ec8b116a5eff450fcde.png]  
在`Producer`类中，分别发送aaa、bbb、ccc、ddd四个消息：  
![在这里插入图片描述][4be51e3c72904a2b954984a37ee96073.png]  
第一个消费者，收到了aaa、ccc：  
![在这里插入图片描述][d7c9b7e86acb47c4938be0effe5f83e3.png]  
第二个消费者，收到了bbb、ddd：  
![在这里插入图片描述][fd29d82e1b4c41c6998bd95ce625fe10.png]  
**总结**：在该模式下，一个消息只能被一个线程消费，采用的是轮询接收。

### 发布/订阅模式 ###

在`发布/订阅模式`中，消息的发送者通过消息通道广播出去，让订阅改消息主题的订阅者消费。

就是一个生产者发送的消息会被多个消费者获取，所以也叫广播模式、一对多模式。  
![在这里插入图片描述][72a77374bb5e42188b9b1735f11d95c4.png]  
**流程示意图如下**：

1.  生产者将消息发送到交换机
2.  交换机将信息发给所有绑定的队列
3.  绑定队列的消费者收到消息

该模式需要指定一个`Exchange`交换机，起本身只负责转发消息，不具备存储消息的能力。一方面，接收生产者发送的消息。另一方面，需要知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。

到底如何操作，取决于`Exchange`的类型。有常见以下几种类型：

*  **Fanout**：广播。将消息交给所有绑定到交换机的队列
 *  **Direct**：定向。把消息交给符合指定routing key的队列
 *  **Topic**：通配符。把消息交给符合routing pattern(路由模式)的队列
 *  **Header**：通过消息内容中的headers属性来进行匹配

按照流程，编写生产者代码：

public class Producer {
        
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 1. 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 2. 新建连接
            Connection connection = factory.newConnection();
            // 3. 新建通道
            Channel channel = connection.createChannel();
            /*
             * 4. 声明一个交换机
             * exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean , Map<String, Object> arguments)
             *  参数：
             *  exchange=》交换机名称
             *  type=》指定交换机的类型为FANOUT广播模式
             *  durable=》是否持久化
             *  autoDelete=》自动删除
             *  internal=》内部使用，一般为false
             *  arguments=》参数
             */
            String exchangeName = "fanoutExchange";
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT, true, false, false, null);
            // 5. 创建队列
            String fanoutQueueOne = "fanoutQueueOne";
            String fanoutQueueTwo = "fanoutQueueTwo";
            channel.queueDeclare(fanoutQueueOne, true, false, false, null);
            channel.queueDeclare(fanoutQueueTwo, true, false, false, null);
            // 6. 绑定队列和交换机,参数（队列名称、交换机名称，路由键（绑定规则，如果交换机的类型为fanout，routingKey为“”）
            channel.queueBind(fanoutQueueOne, exchangeName, "");
            channel.queueBind(fanoutQueueTwo, exchangeName, "");
            // 7. 发送消息
            String body = "广播消息";
            channel.basicPublish(exchangeName, "", null, body.getBytes(StandardCharsets.UTF_8));
            // 8. 关闭连接
            channel.close();
            connection.close();
            System.out.println("消息发送完毕");
        }
    }

运行程序，发送消息，在控制台中，可以看到创建的交换机、绑定关系如下图：  
![在这里插入图片描述][b8b8f0a7cff24b68b941a9f6cfd9f0ad.png]  
在创建的两个队列中，可以看到都收到了消息，并处于就绪状态：  
![在这里插入图片描述][ca50c8f3af6c47b4b3cfdb1629e72787.png]

**创建消费者**，并复制代码，运行两个消费者：

public class FanoutConsumerOne {
        
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 1. 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 2. 创建连接及通道
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();   //与rabbitmq通讯的通道
            // 3. 绑定队列，队列和交换机绑定
            String fanoutQueueOne = "fanoutQueueOne";
            String exchangeName = "fanoutExchange";
            channel.queueBind(fanoutQueueOne, exchangeName, "");
            // 4. 处理消息的回调对象
            DeliverCallback deliverCallback = (consumerTag, delivery) ->
            {
        
                String message = new String(delivery.getBody());
                System.out.println(message);
            };
            // 4. 取消消息处理的回调对象
            CancelCallback cancelCallback = s -> {
        
                System.out.println("消息消费被中断");
            };
            // 5. 接收消息，把消息传递到回调对象处理
            channel.basicConsume(fanoutQueueOne, true, deliverCallback, cancelCallback);
        }
    }

可以看到，每个消息都会被所有消费者接收：  
![在这里插入图片描述][d16bfd1c75e8474795f88b8c78e0a71a.png]

![在这里插入图片描述][be385642eb604d10a9b126e45990baf9.png]

### 路由模式 ###

官网的**路由（Routing）模式**流程示意图如下：  
![在这里插入图片描述][648b7a5bf4254924b126461871c50744.png]  
该模式也需要加入**交换机**，指定其类型为`direct`。队列在绑定交换机时要指定`routing key（路由键）`，消息会转发到符合`routing key`的队列。交换机根据`routingKey`进行完全匹配，如果匹配失败则丢弃消息。

**流程说明**：

1.  生产者**P**，向**Exchange**发送消息，发送消息时，会指定一个 **routing key**。
2.  **交换机**接收生产者的消息，然后把消息递交给与 **routing key** 完全匹配的队列。
3.  消费者**C1**，其所在队列指定了需要 **routing key** 为 error 的消息。
4.  消费者**C2**，其所在队列指定了需要 **routing key** 为 info、error、warning 的消息。

**生产者核心代码**：

// 4. 声明一个交换机
            String exchangeName = "directExchange";
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true, false, false, null);
            // 5. 创建队列
            String directQueueInfo = "directQueueInfo";
            String directQueueError = "directQueueError";
            channel.queueDeclare(directQueueInfo, true, false, false, null);
            channel.queueDeclare(directQueueError, true, false, false, null);
            // 6. 绑定队列和交换机,参数（队列名称、交换机名称，路由键）
            channel.queueBind(directQueueInfo, exchangeName, "info");
            channel.queueBind(directQueueError, exchangeName, "error");
            // 7. 发送消息
            channel.basicPublish(exchangeName, "info", null, "INFO日志".getBytes(StandardCharsets.UTF_8));
            channel.basicPublish(exchangeName, "error", null, "ERROR日志".getBytes(StandardCharsets.UTF_8));

**消费者代码**：

// 3. 绑定队列
            String directQueueError = "directQueueError";
            // 队列和交换机绑定
            String exchangeName = "directExchange";
            channel.queueBind(directQueueError, exchangeName, "error");
            // 4. 处理消息的回调对象
            DeliverCallback deliverCallback = (consumerTag, delivery) ->
            {
        
                String message = new String(delivery.getBody());
                System.out.println(message);
            };
            // 4. 取消消息处理的回调对象
            CancelCallback cancelCallback = s -> {
        
                System.out.println("消息消费被中断");
            };
            // 5. 接收消息，把消息传递到回调对象处理，参数：
            channel.basicConsume(directQueueError, true, deliverCallback, cancelCallback);

再复制代码创建另外一个消费者，绑定路由键为`info`，分别启动后运行发送消息。

结果如下所示，绑定了`error`路由的队列只收到ERROR日志，`info`路由的队列只收到INFO日志。  
![在这里插入图片描述][ecb917905b9b4c789de07bbc201201ef.png]![在这里插入图片描述][997d3649a6694221b57833bf7a977367.png]

### 主题模式 ###

![在这里插入图片描述][fd81839d21604ef5ab29955b078c4576.png]

主题`Topic`类型与`Direct`相比，都是可以根据`RoutingKe`y把消息路由到不同的队列。只不过`Topic`类型的`Routing key` 可以使用**通配符**。

`Routingkey`一般都是有一个或多个单词组成，多个单词之间以`”.”`分割，例如：`order.insert`

**通配符规则**：

*  `#`：匹配一个或多个词
 *  `*`：匹配一个词

**举例**：

*  `item.#`：能够匹配`item.spu.insert` 或者`item.spu`
 *  `item.*`：只能匹配`item.spu`

创建 `Topic`类型的交换机，发送消息：

// 4. 声明一个交换机类型为TOPIC
            String exchangeName = "topicExchange";
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC, true, false, false, null);
            // 5. 发送消息时，携带通配符KEY
            channel.basicPublish(exchangeName, "aa.error", null, "日志=》aa.error".getBytes(StandardCharsets.UTF_8));
            channel.basicPublish(exchangeName, "bbb.ccc.error", null, "日志=>bbb.ccc.error".getBytes(StandardCharsets.UTF_8));

创建两个消费者，使用通配符绑定：

// 3. 绑定队列
            String topicQueueError = "topicQueueError";
            // 队列和交换机绑定,*.error 只能接到aa.error、bb.error等前缀为一个单词的消息
            String exchangeName = "topicExchange";
            channel.queueBind(topicQueueError, exchangeName, "*.error");

// 3. 绑定队列
            String topicQueueError = "topicQueueError";
            // 队列和交换机绑定,#.error 只能接到aa.error、bb.cc.error等前缀为一个或多个单词的消息
            String exchangeName = "topicExchange";
            channel.queueBind(topicQueueError, exchangeName, "#.error");

运行代码，发送消息，运行结果如下：  
![在这里插入图片描述][3a366a2f516d44bd9586441cf8c0004a.png]

![在这里插入图片描述][b6e3dc9102c24b4082ba5f1be0cba150.png]

### RPC 模式 ###

`RPC 模式`也就是使用队列实现远程过程调用。实际使用的很少，专业的`RPC`框架已经很多了~，这里就不介绍了。  
![在这里插入图片描述][a7864e70a2914864a7916f77932c3b6c.png]

[7f602f427192406aad665541ca1e7dc3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/81411ddf55ee4d8e9f27543e3c7e26a5.png
[2e315c569a8c483aa2aa256854a89f37.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/35f1e70231d2479ab54c341c9de85c23.png
[RabbitMQ Java Client GitHub]: https://github.com/rabbitmq/rabbitmq-java-client
[d64ab0145dbe47d19e137fcd5c6955c9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/4dd6ce92f34a4de8bc51cbe0db6d5ff8.png
[27a152371a774116a23d998b97ccbc11.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/1e7ad7c01fda4978a5e01183b8363911.png
[0ac513f7b2e14cad9ab5638653ca4619.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/1bd05b4c30fe4bd59821a971951758b7.png
[195ad863d09d4ec8b116a5eff450fcde.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/7edb10ce2fcc45279e480a9caa8f2647.png
[4be51e3c72904a2b954984a37ee96073.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/c334d822189e44fe957e388e9354078b.png
[d7c9b7e86acb47c4938be0effe5f83e3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/7aa7b02b6429457990cd3845253ccb75.png
[fd29d82e1b4c41c6998bd95ce625fe10.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/02bc84b1a8be4e96b1155bc178d7ec1f.png
[72a77374bb5e42188b9b1735f11d95c4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/c8c42855fcdd425fb82f2c7b0ea65c46.png
[b8b8f0a7cff24b68b941a9f6cfd9f0ad.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/d92be586f6844733adae68ce80a9a56c.png
[ca50c8f3af6c47b4b3cfdb1629e72787.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/7dca672080a54522a688c208b9453235.png
[d16bfd1c75e8474795f88b8c78e0a71a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/7469fd22e555418388bd0a80adbd3c9e.png
[be385642eb604d10a9b126e45990baf9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/c3dd865b18c342f184d599de47f244e4.png
[648b7a5bf4254924b126461871c50744.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/7b474935da4d453398af73ef808bd411.png
[ecb917905b9b4c789de07bbc201201ef.png]: https://img-blog.csdnimg.cn/ecb917905b9b4c789de07bbc201201ef.png
[997d3649a6694221b57833bf7a977367.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/c883b94d24284989a10600ead89d4585.png
[fd81839d21604ef5ab29955b078c4576.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/6378ef6f022b4946996f1d5343c98b51.png
[3a366a2f516d44bd9586441cf8c0004a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/6e5c0cd1847e426cb21b0b1025fccf23.png
[b6e3dc9102c24b4082ba5f1be0cba150.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/327d46159b984c3d8b85e46a40e1abce.png
[a7864e70a2914864a7916f77932c3b6c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/2958d550a2d041d088d029441dc2ef8a.png