RabbitMQ--消息分发、消息顺序性、消息传输保障

绝地灬酷狼 2021-08-29 20:35 466阅读 0赞

### 消息分发 ###

当RabbitMQ队列拥有多个消费者时，队列收到的消息将以轮询(round-robin) 的分发方式发送给消费者。每条消息只会发送给订阅列表里的一个消费者。这种方式非常适合扩展，而且它是专门为并发程序设计的。如果现在负载加重，那么只需要创建更多的消费者来消费处理消息即可。

很多时候轮询的分发机制也不是那么优雅。默认情况下,如果有n个消费者,那么RabbitMQ会将第m条消息分发给第m%n (取余的方式)个消费者，RabbitMQ 不管消费者是否消费并已经确认(Basic.Ack)了消息。试想一下，如果某些消费者任务繁重，来不及消费那么多的消息，而某些其他消费者由于某些原因(比如业务逻辑简单、机器性能卓越等)很快地处理完了所分配到的消息，进而进程空闲，这样就会造成整体应用吞吐量的下降。

那么该如何处理这种情况呢?这里就要用到`channel . basicQos (int prefetchCount)`这个方法，channel . basicQos方法允许限制信道上的消费者所能保持的最大未确认消息的数量。

举例说明，在订阅消费队列之前，消费端程序调用了channel. basicQos(5)，之后订阅了某个队列进行消费。RabbitMQ会保存一个消费者的列表,每发送一条消息都会为对应的消费者计数，如果达到了所设定的上限，那么RabbitMQ就不会向这个消费者再发送任何消息。直到消费者确认了某条消息之后,RabbitMQ将相应的计数减1,之,后消费者可以继续接收消息，直到再次到达计数上限。这种机制可以类比于TCP/IP中的“ 滑动窗口”。

注意：Basic.Qos的使用对于拉模式的消费方式无效。

(1) void basicQos(int prefetchCount) throws IOException;
    
    (2) void basicQos(int prefetchCount, boolean global) throws IOException;
    
    (3) void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

前面介绍的都只用到了prefetchCount这个参数,当prefetchCount设置为0则表示没有上限。还有prefetchSize这个参数表示消费者所能接收未确认消息的总体大小的上限，单位为B，设置为0则表示没有，上限。

对于一个信道来说，它可以同时消费多个队列，当设置了prefetchCount大于0时，这个信道需要和各个队列协调以确保发送的消息都没有超过所限定的prefetchCount的值，这样会使RabbitMQ的性能降低，尤其是这些队列分散在集群中的多个Broker节点之中。RabbitMQ 为了提升相关的性能，在AMQP0-9-1协议之上重新定义了global这个参数，对比如下表所示。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

对于同一个信道上的多个消费者而言，如果设置了prefetchCount的值，那么都会生效。

//两个消费者，各自能接收到的未确认消息的上限都为10
    Consumer consumer1 = new DefaultConsumer(channel);
    Consumer consumer2 = new DefaultConsumer(channel);
    channel.basicQos(10);//Per consumer limit
    channel.basicConsume("my-queue1", false, consumer1);
    channel.basicConsume("my-queue1", false, consumer2);

如果在订阅消息之前，既设置了global为true的限制，又设置了global 为false的限制，那么哪个会生效呢? RabbitMQ会确保两者都会生效。举例说明，当前有两个队列queuel和queue2:queuel有10条消息，分别为1到10;queue2也有10条消息，分别为11到20。有两个消费者分别消费这两个队列:

//两个消费者，各自能接收到的未确认消息的上限都为10
    Consumer consumer1 = new DefaultConsumer(channel);
    Consumer consumer2 = new DefaultConsumer(channel);
    channel.basicQos(3, false);//Per consumer limit
    channel.basicQos(5, true);//Per channel limit
    channel.basicConsume("my-queue1", false, consumer1);
    channel.basicConsume("my-queue1", false, consumer2);

`那么这里每个消费者最多只能收到3个未确认的消息，两个消费者能收到的未确认的消息个数之和的上限为5`。在未确认消息的情况下，如果consumer1接收到了消息1、2和3，那么consumer2至多只能收到11和12。如果像这样同时使用两种global的模式，则会增加RabbitMQ的负载，因为RabbitMQ需要更多的资源来协调完成这些限制。如无特殊需要，最好只使用global为false的设置，这也是默认的设置。

#### 消息顺序性 ####

消息的顺序性是指消费者消费到的消息和发送者发布的消息的顺序是一致的。举个例子，不考虑消息重复的情况，如果生产者发布的消息分别为msg1、msg2、msg3,那么消费者必然也是按照msg1、msg2、msg3 的顺序进行消费的。

目前很多资料显示RabbitMQ的消息能够保障顺序性，这是不正确的，或者说这个观点有很大的局限性。在不使用任何RabbitMQ的高级特性，也没有消息丢失、网络故障之类异常的情况发生，并且只有一个消费者的情况下，最好也只有一个生产者的情况下可以保证消息的顺序性。如果有多个生产者同时发送消息，无法确定消息到达Broker的前后顺序，也就无法验证消息的顺序性。

那么哪些情况下RabbitMQ的消息顺序性会被打破呢?下面介绍几种常见的情形。

如果生产者使用了事务机制，在发送消息之后遇到异常进行了事务回滚，那么需要重新补偿发送这条消息，如果补偿发送是在另一个线程实现的，那么消息在生产者这个源头就出现了错序。同样，如果启用publisher confirm时，在发生超时、中断，又或者是收到RabbitMQ的Basic.Nack命令时，那么同样需要补偿发送，结果与事务机制一样会错序。或者这种说法有些牵强，我们可以固执地认为消息的顺序性保障是从存入队列之后开始的，而不是在发送的时候开始。

考虑另一种情形,如果生产者发送的消息设置了不同的超时时间，并且也设置了死信队列，整体上来说相当于一个延迟队列，那么消费者在消费这个延迟队列的时候，消息的顺序必然不会和生产者发送消息的顺序一致。

再考虑一种情形,如果消息设置了优先级,那么消费者消费到的消息也必然不是顺序性的。如果一个队列按照前后顺序分有msg1、msg2、msg3、msg4这4个消息，同时有ConsumerA和ConsumerB这两个消费者同时订阅了这个队列。队列中的消息轮询分发到各个消费者之中，ConsumerA中的消息为msg1和msg3，ConsumerB 中的消息为msg2、msg4。ConsumerA 收到消息msg1之后并不想处理而调用了Basic.Nack/.Reject将消息拒绝，与此同时将requeue设置为true， 这样这条消息就可以重新存入队列中。消息msg1之后被发送到了ConsumerB中，此时ConsumerB已经消费了msg2、 msg4，之,后再消费msgl，这样消息顺序性也就错乱了。或者消息msg1又重新发往ConsumerA中，此时ConsumerA已经消费了msg3,那么再消费msg1，消息顺序性也无法得到保障。同样可以用在Basic. Recover这个AMQP命令中。

包括但不仅限于以上几种情形会使RabbitMQ 消息错序。如果要保证消息的顺序性，需要业务方使用RabbitMQ之后做进一步的处理,比如在消息体内添加全局有序标识(类似SequenceID)来实现。

弃用的QueueingConsumer:

//弃用的QueueConsumer
    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicQos(64);//使用QueueingConsumer的时候一定要添加
    channel.basicConsume(QUEUE_NAME, false, "consumer_zzh", consumer);
    while (true) { 
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        String message = new String(delivery.getBody());
        System.out.println("[X] Rceived '" + message + "'");
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }

。QueueingConsumer本身有几个大缺陷，需要读者在使用时特别注意。首当其冲的就是内存溢出的问题，如果由于某些原因，队列之中堆积了比较多的消息，就可能导致消费者客户端内存溢出假死，于是发生恶性循环，队列消息不断堆积而得不到消化。

采用上面的代码进行演示，首先向一个队列发送200多MB的消息，然后进行消费。在客户端调用channel.basicConsume方法订阅队列的时候，RabbitMQ 会持续地将消息发往QueueingConsumer中,QueueingConsumer内部使用LinkedBlockingQueue来缓存这些消息。通过JVisualVM可以看到堆内存的变化，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

可以看到堆内存一直在增加， 这里只测试了发送200MB左右的消息，如果发送更多的消息，那么这个堆内存会变得更大，直到出现java. lang . OutOfMemoryError的报错。  
这个内存溢出的问题可以使用Basic. Qos来得到有效的解决，Basic. Qos可以限制某个消费者所保持未确认消息的数量，也就是间接地限制了QueueingConsumer 中的LinkedBlockingQueue的大小。注意一定 要在调用Basic. Consume之前调用Basic. Qos才能生效。

Queue ingConsumer还包含(但不仅限于)以下一些缺陷:

*  QueueingConsumer 会拖累同一个Connection下的所有信道，使其性能降低;
 *  同步 递归调用QueueingConsumer会产生死锁;
 *  RabbitMQ的自动连接恢复机制( automatic connection recovery) 不支持QueueingConsumer的这种形式;
 *  QueueingConsumer 不是事件驱动的。

为了避免不必要的麻烦，建议在消费的时候尽量使用继承DefaultConsumer的方式。

#### 消息传输保障 ####

消息可靠传输一般是业务系统接入消息中间件时首要考虑的问题，一般消息中间件的消息传输保障分为三个层级。

*  At most once:最多一次。 消息可能会丢失，但绝不会重复传输。
 *  At least once:最少一次。消息绝不会丢失，但可能会重复传输。
 *  Exactly once:恰好一次。每条消息肯定会被传输一次且仅传输一次。

RabbitMQ支持其中的“最多一次”和“最少一次”。其中“最少一次”投递实现需要考虑以下这个几个方面的内容:

1.  消息生产者需要开启事务机制或者publisher confirm机制，以确保消息可以可靠地传输到RabbitMQ中。
2.  消息生产者需要配合使用mandatory参数或者备份交换器来确保消息能够从交换器路由到队列中，进而能够保存下来而不会被丢弃。
3.  消息和队列都需要进行持久化处理，以确保RabbitMQ服务器在遇到异常情况时不会造成消息丢失。
4.  消费者在消费消息的同时需要将autoAck设置为false, 然后通过手动确认的方式去确认已经正确消费的消息，以避免在消费端引起不必要的消息丢失。

“最多一次”的方式就无须考虑以上那些方面，生产者随意发送，消费者随意消费，不过这样很难确保消息不会丢失。

“恰好一次”是RabbitMQ目前无法保障的。考虑这样一种情况，消费者在消费完一条消息之后向RabbitMQ发送确认Basic.Ack命令，此时由于网络断开或者其他原因造成RabbitMQ并没有收到这个确认命令，那么RabbitMQ 不会将此条消息标记删除。在重新建立连接之后，消费者还是会消费到这一条消息，这就造成了重复消费。再考虑一种情况，生产者在使用publisher confirm机制的时候，发送完一条消息等待RabbitMQ返回确认通知，此时网络断开,生产者捕获到异常情况，为了确保消息可靠性选择重新发送，这样RabbitMQ中就有两条同样的消息，在消费的时候，消费者就会重复消费。

那么RabbitMQ有没有去重的机制来保证“恰好一次”呢?答案是并没有,不仅是RabbitMQ, .目前大多数主流的消息中间件都没有消息去重机制，也不保障“恰好一次”。去重处理一般是在业务客户端实现，比如引入GUID (Globally Unique Identifier)的概念。针对GUID，如果从客户端的角度去重，那么需要引入集中式缓存，必然会增加依赖复杂度，另外缓存的大小也难以界定。建议在实际生产环境中，业务方根据自身的业务特性进行去重，比如业务消息本身具备幂等性，或者借助Redis等其他产品进行去重处理。

**总结：**

提升数据可靠性有以下一些途径:

*  设置mandatory参数或者备份交换器( immediate参数已被淘汰);
 *  设置publisher confirm机制或者事务机制;
 *  设置交换器、队列和消息都为持久化;
 *  设置消费端对应的 toA :参数为false并在消费完消息之后再进行消息确认。

本章不仅介绍了数据可靠性的一-些细节，还展示了RabbitMQ 的几种已具备或者衍生的高级特性，包括TTL、死信队列、延迟队列、优先级队列、RPC功能等，这些功能在实际使用中可以让相应应用的实现变得事半功倍。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/d753a8a2a89148af97b7fafe6c0958dd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/4ef9d0901e1c4142b524f0bf5956d45b.png