RabbitMQ的四种转换器模式

柔情只为你懂 2022-05-19 13:11 245阅读 0赞

> # **RabbitMQ的四种转换器模式总结**  #

**【参考文献】：[（1）][1]、[（2）][2]**

### **1. 工作队列：** ###

**（一）Round-robin 转发机制：一次性全部转发，一个一个的发送信息给下一个消费者(consumer)，而不考虑每个任务的时长等等，且是一次性分配，并非一个一个分配。平均的每个消费者将会获得相等数量的消息。**

【发送方（生产者）】：
    /**
     * 第一个参数：queue QUEUE_NAME :队列名称 
     * 第二个参数：durable durable : false 队列消息默认不能持久化，（true）在服务器重启时，能够存活
     * 第三个参数：exclusive : true  => 队列是专用队列 ，是否为当前连接的专用队列，在连接断开后，会自动删除该队列（生产少用）
     * 第四个参数：autoDelete : 当没有任何消费者使用时，自动删除该队列
     * 第五个参数：Map<String, Object> arguments ：队列的其他参数，配置属性
     */
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    /**
     * 第一个参数：exchange 转换器转发消息功能 
     * 第二个参数：routingKey 路由消息携带的选择键 （这里是队列名称QUEUE_NAME）
     * 第三个参数：BasicProperties props 消息配置属性，例如 rounting headers模式等
     * 第四个参数：byte[] body 消息内容
     */
    channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    
    【接收方（消费者）】：
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    QueueingConsumer consumer = new QueueingConsumer(channel);
    
    /**
     * 第一个参数：queue 队列名称 
     * 第二个参数：autoAck  false ：打开应答机制
     * 第三个参数：Consumer callback 消费者对象
     */
    // 指定消费队列
    channel.basicConsume(QUEUE_NAME, true, consumer);

**（二）消息应答 **

**消息应答（message acknowledgments）：**

**【目的】：当某个工作者（接收者）被杀死时，我们希望将任务传递给另一个工作者。为了保证消息永远不会丢失。**

**【方案】：消费者发送应答给RabbitMQ，告诉它信息已经被接收和处理，然后RabbitMQ可以自由的进行信息删除。如果消费者被杀死而没有发送应答，RabbitMQ会认为该信息没有被完全的处理，然后将会重新转发给别的消费者。**

**注：这种机制并没有超时时间这么一说，RabbitMQ只有在消费者连接断开是重新转发此信息。如果消费者处理一个信息需要耗费特别特别长的时间是允许的**

【接收方（消费者）】：
    // 打开应答机制
    boolean ack = false ;
    channel.basicConsume(QUEUE_NAME, ack, consumer);
    
    //另外需要在每次处理完成一个消息后，手动发送一次应答
    /**
     * 第一个参数：deliveryTag: 该消息的index
     * 第二个参数：multiple：   是否批量.true:将一次性ack所有小于deliveryTag的消息。
     */
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

**（三）消息持久化（Message durability）**

**【目的】：RabbitMQ退出或者异常退出，确保信息不会被丢失。**

**【方案】：需要给所有的队列和消息设置持久化的标志。**

【发送方（生产者）】:
    /**
     * 第一， 我们需要确认RabbitMQ永远不会丢失我们的队列。为了这样，我们需要声明它为持久化的。
     */
    boolean durable = true;
    //声明队列
    channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
    
    /**
     * 第二， 我们需要标识我们的信息为持久化的。通过设置MessageProperties（implements BasicProperties）
     * 值为PERSISTENT_TEXT_PLAIN
     * 第三个参数：BasicProperties 需要注意的是BasicProperties.deliveryMode，0:不持久化 1：持久化 *这里指的是消息的持久化，
     * 配合channel(durable=true),queue(durable)可以实现，即使服务器宕机，消息仍然保留
     */
    channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

**注： RabbitMQ不允许使用不同的参数重新定义一个队列，所以已经存在的队列，我们无法修改其属性。**

**（四）公平转发（Fair dispatch）：一个一个的分配 区别于 Round-robin 转发机制**

![20171212095542074][]

**【区别】： **

**当消费者不忙时进行转发。且这种模式下支持动态增加消费者，因为消息并没有发送出去，动态增加了消费者马上投入工作。而默认的转发机制（Round-robin转发机制）会造成，即使动态增加了消费者，此时的消息已经分配完毕，无法立即加入工作，即使有很多未完成的任务。**

**【目的】：**

**       存在的现象：例如，这样一种情况，对于两个消费者，有一系列的任务，奇数任务特别耗时，而偶数任务却很轻松，这样造成一个消费者一直繁忙，另一个消费者却很快执行完任务后等待。**

**RabbitMQ仅仅是当消息到达队列进行转发消息。并不在乎有多少任务消费者并未传递一个应答给RabbitMQ。仅仅盲目转发所有的奇数给一个消费者，偶数给另一个消费者。 **

**【方案】：****使用basicQos方法，传递参数为prefetchCount = 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。换句话说，只有在消费者空闲的时候会发送下一条信息。**

【接收方（消费者）】：
     //设置最大服务转发消息数量
     int prefetchCount = 1;
     channel.basicQos(prefetchCount);

### **2.转换器（Exchanges） ** ###

**RabbitMQ消息模型的核心理念是生产者永远不会直接发送任何消息给队列，一般的情况生产者甚至不知道消息应该发送到哪些队列。**

![20171212095628115][]

**P：生产者；X：exchange转发器 ；队列**

**转发器的类型：Direct、Topic、Headers、Fanout**

--------------------

**介绍四种转发器模式前先介绍几个概念： **

**（1）匿名转发器（nameless exchange）：**

**前面说到生产者只能发送消息给转发器（Exchange），但是我们前面的例子并没有使用到转发器，我们仍然可以发送和接收消息。这是因为我们使用了一个默认的转发器，它的标识符为 ""（下面第一个参数）。**

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

**（2）临时队列（Temporary queues）：**

**【背景】：**

**对于我们的日志系统我们并不关心队列的名称。我们想要接收到所有的消息，而且我们也只对当前正在传递的数据的感兴趣。**

**【条件】：**

**（1）无论什么时间连接到Rabbit我们都需要一个新的空的队列。为了实现，我们可以使用随机数创建队列，或者更好的，让服务器给我们提供一个随机的名称。**

**（2）一旦消费者与Rabbit断开，消费者所接收的那个队列应该被自动删除。**

// Java中我们可以使用queueDeclare()方法，不传递任何参数，来创建一个非持久的、唯一的、自动删除的队列且队列名称由服务器随机产生。
    【接收方（消费者）】：
    String queueName = channel.queueDeclare().getQueue();
    // 一般情况这个名称与amq.gen-JzTY20BRgKO-HjmUJj0wLg 类似。

**（3） 绑定（Bindings）**

![20171212095848604][]

【接收方（消费者）】：
    /**
     * 第一个参数：queue 队列名称
     * 第二个参数：exchange 转发器名称
     * 第三个参数：routingKey 绑定键 （在转发器与队列之间）
     */
    // 用于通过绑定bindingKey将queue到Exchange，之后便可以进行消息接收
    channel.queueBind(queueName, EXCHANGE_NAME, "");

--------------------

**（一）fanout 转发器**

**【定义】：fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。**

![20171212100117880][]

**上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。 **

【发送方（生产者）】:
    // 声明转发器和类型
    /**
    * 第一个参数：exchange 转发器名称
    * 第二个参数：type 转发器类型
    */
    channel.exchangeDeclare(EXCHANGE_NAME, "fanout" );
    
    // 往转发器上发送消息
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
    
    【接收方（消费者）】：
    channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    // 创建一个非持久的、唯一的且自动删除的队列
    String queueName = channel.queueDeclare().getQueue();
    // 为转发器指定队列，设置binding
    channel.queueBind(queueName, EXCHANGE_NAME, "");
    // 绑定消费者到转换器
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 指定接收者，第二个参数为自动应答，无需手动应答
    /**
    * 第二个参数：autoAck：是否自动ack，
    * 如果不自动ack，需要使用channel.ack、channel.nack、channel.basicReject 进行消息应答
    */
    channel.basicConsume(queueName, true, consumer);

**（二）direct 转发器**

**【定义】：direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。 **

![20171212100205443][]

**以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。 **

**【背景】： **

** 我们可能希望把致命类型的错误写入硬盘，而不把硬盘空间浪费在警告或者消息类型的日志上。来允许根据日志的严重性进行过滤日志。**

【发送方（生产者）】:
    // 声明转发器的类型
    channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
    // 发布消息至转发器，指定routingkey
    channel.basicPublish(EXCHANGE_NAME, routingKey, null, message
    .getBytes());
    
    
    【接收方（消费者）】：
    // 声明direct类型转发器
    channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
    String queueName = channel.queueDeclare().getQueue();
    // 指定binding_key
    channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
    
    QueueingConsumer consumer = new QueueingConsumer(channel);
    
    channel.basicConsume(queueName, true, consumer);

** 注：发送方（生产者）支持发送消息给多个转发器：**

package com.caox.rabbitmq.demo._04_binding_key;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.util.Random;
    import java.util.UUID;
    
    /**
     * Created by nazi on 2018/7/25.
     * Direct转换器 路由选择
     */
    public class EmitLogDirectProducer {
        private static final String EXCHANGE_NAME = "ex_logs_direct";
        private static final String EXCHANGE_NAME_ALL = "ex_logs_direct_all";
        private static final String[] SEVERITIES = { "info", "warning", "error" };
    
        public static void main(String[] argv) throws Exception {
            // 创建连接和频道
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明转发器的类型
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
            channel.exchangeDeclare(EXCHANGE_NAME_ALL,"fanout");
            //发送6条消息
            for (int i = 0; i < 6; i++)
            {
                String severity = getSeverity();
                String message = severity + "_log :" + UUID.randomUUID().toString();
                // 发布消息至转发器，指定routingkey
                channel.basicPublish(EXCHANGE_NAME, severity, null, message
                        .getBytes());
    
                channel.basicPublish(EXCHANGE_NAME_ALL, severity, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
            }
    
            channel.close();
            connection.close();
        }
    
        /**
         * 随机产生一种日志类型
         *
         * @return
         */
        private static String getSeverity()
        {
            Random random = new Random();
            int ranVal = random.nextInt(3);
            return SEVERITIES[ranVal];
        }
    }

**注：多重绑定（multiple bindings）：**

![Center][]

**使用一个绑定键（binding key）绑定多个队列是完全合法的。如上图，一个附带选择键（routing key）的消息将会被转发到Q1和Q2。**

**（三）topic 转发器**

**topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：**

**routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”**

**注意：你可以定义任何数量的标识符，上限为255个字节。**

**binding key与routing key一样也是句点号“. ”分隔的字符串。**

**binding key中可以存在两种特殊字符“\*”与“\#”，用于做模糊匹配，其中“\*”用于匹配一个单词，“\#”用于匹配多个单词（可以是零个）。**

![20171212100341210][]

**以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。**

**注：主题类型的转发器非常强大，可以实现其他类型的转发器。  
       当一个队列与绑定键\#绑定，将会收到所有的消息，类似fanout类型转发器。  
       当绑定键中不包含任何\#与\*时，类似direct类型转发器。 **

【发送方（生产者）】:
    // 声明转发器的类型
    channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    // 发布消息至转发器，指定routingkey
    channel.basicPublish(EXCHANGE_NAME, routing_key, null, msg
    .getBytes());
    
    
    【接收方（消费者）】：
    // 声明转发器
    channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    // 随机生成一个队列
    String queueName = channel.queueDeclare().getQueue();
    
    // 接收所有与kernel相关的消息
    channel.queueBind(queueName, EXCHANGE_NAME, "*.critical");
    
    QueueingConsumer consumer = new QueueingConsumer(channel);
    
    channel.basicConsume(queueName, true, consumer);

**（四）headers 转发器**

**headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。**

**在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。**

【发送方（生产者）】：
    //声明转发器和类型headers
    /**
    * 第一个参数：exchange 转发器名称
    * 第二个参数：type 转发器类型  type有direct、fanout、topic、headers四种。
    * 第三个参数：durable：true、false；true：服务器重启会保留下来Exchange。
    *			  警告：仅设置此选项，不代表消息持久化。即不保证重启后消息还在。
    * 第四个参数：autoDelete：true、false.true:当已经没有消费者时，服务器是否可以删除该Exchange。
    * 第五个参数：Map<String, Object> arguments：转换器的其他属性。
    */
    channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS, false,true,null);
    
    Map<String,Object> headers =  new Hashtable<String, Object>();
    headers.put("aaa", "01234");
    headers.put("abc","key");
    AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties.Builder();
    properties.headers(headers);
    
    // 指定消息发送到的转发器,绑定键值对headers键值对
    channel.basicPublish(EXCHANGE_NAME, "",properties.build(),message.getBytes());
    
    
    【接收方（消费者）】：
    //声明转发器和类型headers
    channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS,false,true,null);
    
    channel.queueDeclare(QUEUE_NAME,false, false, true,null);
    
    Map<String, Object> headers = new Hashtable<String, Object>();
    //all any
    headers.put("x-match", "any");
    headers.put("abc", "01234");
    
    // 为转发器指定队列，设置binding 绑定header键值对
    /**
    * 第四个参数：Map<String, Object> arguments：绑定参数
    */
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME,"", headers);
    
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 指定接收者，第二个参数为自动应答，无需手动应答
    channel.basicConsume(QUEUE_NAME, true, consumer);

**注意：**

**（1）fanout，direct，topic exchange的routingKey都需要要字符串形式的，而headers exchange则没有这个要求，因为键值对的值可以是任何类型 ，匹配有两种方式all和any。这两种方式是在接收端必须要用键值"x-mactch"来定义。all代表定义的多个键值对都要满足，而any则代码只要满足一个就可以了。**

**（2） x-patch: all 满足 生产者和消费者的 参数要一模一样（个数和内容），多一个少一个都不行。**

## **3.远程过程调用（RPC） ** ##

**MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。**

**但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。**

### ** RPC工作流程:** ###

![20171212100410528][]

**1)、客户端启动时，创建了一个匿名的回调队列。  
2)、在一个RPC请求中，客户端发送一个消息，它有两个属性：1.REPLYTO，用来设置回调队列名；2.correlationId，对于每个请求都被设置成唯一的值。  
3)、请求被发送到rpc\_queue队列.  
4)、RPC工作者（又名：服务器）等待接收该队列的请求。当收到一个请求，它就会处理并把结果发送给客户端，使用的队列是replyTo字段指定的。  
5)、客户端等待接收回调队列中的数据。当接到一个消息，它会检查它的correlationId属性。如果它和设置的相匹配，就会把响应返回给应用程序。**

### **RabbitMQ中实现RPC的机制是：** ###

**客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）；**

**服务器端收到消息并处理；**

**服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性；**

**客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理。**

[1]: https://blog.csdn.net/lmj623565791/article/category/2386657
[2]: https://blog.csdn.net/column/details/slimina-rabbitmq.html?&page=2
[20171212095542074]: /images/20220519/8d1f721e71fd46b9999e33e457c83b90.png
[20171212095628115]: /images/20220519/6fd948e3f6aa45cf99bb0c83ca088eb0.png
[20171212095848604]: /images/20220519/f0c1de1cb4f94ee7af25412c97340800.png
[20171212100117880]: /images/20220519/dec5a091833344c2adb33c65f1fc7530.png
[20171212100205443]: /images/20220519/22c1b489a67042d28b9d77ad74750a71.png
[Center]: /images/20220519/971f0bddfc604821b1aa274f88663449.png
[20171212100341210]: /images/20220519/35662f6da91a442b8e6e2497257b7455.png
[20171212100410528]: /images/20220519/dfec9fee73794704bbedf9ccdbafbf50.png