mq如何保证消息顺序性

Posted 2023-03-15 盖丽男

背景

面试的时候，经常会有面试官问道这个问题，发送顺序消息。

讨论

顺序性其实有两方面，一方面要保证Producer发送时是有序的，Consumer接受和处理消息的有序性。
另一面来说，我们也要考虑是需要全局有序还是局部有序就可以。

kafka

kafka的topic是分Partition的，当有多个Partition的时候，消息可能会按照/或者不按照规则被发送到不同的Partition。
Kafka 中的消费是基于拉模式的，即消费者主动向服务端发起请求来拉取消息。 Kakfa 中的消息消费是一个不断轮询的过程，消费者所要做的就是重复地调用 poll () 方法，而 poll () 方法返回的是所订阅主题（或分区）上的一组消息。

全局有序

所以如果想做到全局有序的话，我们可以只设置一个Partition，这样保证发送的有序性，同时，Consumer端也应该保证接收有序并且不要用多线程，单线程处理。

局部有序

刚才说到了一个topic可以有多个Partition，kafka确保每个Partition只能同一个group中的同一个Consumer消费，所以就Consumer来说，可以保证一个Partition的消息顺序消费，然后kafka的Producer端可以根据要发送消息内容，指定Partition Key，Kafka对其进行Hash计算，根据计算结果决定放入哪个Partition。这样Partition Key相同的消息会放在同一个Partition。此时，Partition的数量仍然可以设置多个，提升Topic的整体吞吐量。

消息重试对顺序消息的影响

对于一个有着先后顺序的消息A、B，正常情况下应该是A先发送完成后再发送B，但是在异常情况下，在A发送失败的情况下，B发送成功，而A由于重试机制在B发送完成之后重试发送成功了。这时对于本身顺序为AB的消息顺序变成了BA。

针对这种问题，严格的顺序消费还需要max.in.flight.requests.per.connection参数的支持。

该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高，就会占用越多的内存，同时也会提升吞吐量。把它设为1就可以保证消息是按照发送的顺序写入服务器的。

此外，对于某些业务场景，设置max.in.flight.requests.per.connection=1会严重降低吞吐量，如果放弃使用这种同步重试机制，则可以考虑在消费端增加失败标记的记录，然后用定时任务轮询去重试这些失败的消息并做好监控报警。

rocketmq

rocketmq的消息模式其实和kafka大差不差，rocketmq的Topic分Queue，当有多个Queue的时候，消息可能会按照/或者不按照规则被发送到不同的Queue。一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。
Kafka 中的消费是基于拉模式的，即消费者主动向服务端发起请求来拉取消息。 Kakfa 中的消息消费是一个不断轮询的过程，消费者所要做的就是重复地调用 poll () 方法，而 poll () 方法返回的是所订阅主题（或分区）上的一组消息。

全局有序

同上，一个Topic只设置一个Queue，Producer保证发送有序。Consumer保证接收有序并且不要用多线程，单线程处理。或者使用可以保证处理顺序的线程模型。

局部有序

rockemt的Consumer消费消息有两种形式：Push和Pull，大多数场景使用的是Push模式，在源码中这两种模式分别对应的是DefaultMQPushConsumer类和DefaultMQPullConsumer类。Push模式实际上在内部还是使用的Pull方式实现的，通过Pull不断地轮询Broker获取消息，当不存在新消息时，Broker端会挂起Pull请求，直到有新消息产生才取消挂起，返回新消息。

那我们就可以通过Producer发送有序消息，通过MessageQueueSelector()来指定消息发送到哪个队列，保证局部有序，消费方启动顺序消费，通过new MessageListenerOrderly()来控制。

消息重试对顺序消息的影响

对于Rocketmq来说，一个Consumer只能注册一个Listener，所以一个consumer要么按顺序消息的方式来消费，要么按普通消息的方式来消费。所以，如果一个用户进程要收两种消息，最好使用两个Consumer实例。
从实例代码可以看出，顺序消息处理消息后返回状态跟普通消息也有所不同，失败的话是返回的SUSPEND_CURRENT_QUEUE_A_MOMENT，而不是RECONSUME_LATER。这是因为对于顺序消息，消费失败是不会返回给broker重新投递的（其实即使重发也还是发到这个consumer上，没必要多此一举），而是会放到本地的缓存队列中重新处理。另外两个状态ROLLBACK和COMMIT已经被设置成deprecated了，我们就不关心了。

参考

一文理解Kafka如何保证消息顺序性
RocketMQ源码解析(十二)-顺序消息