背景
面试的时候,经常会有面试官问道这个问题,发送顺序消息。
讨论
顺序性其实有两方面,一方面要保证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
了,我们就不关心了。