RocketMQ消费失败消息深入分析(consumer,broker的具体处理逻辑)

谢柳平

前言

消息队列是分布式系统中重要的组件,主要解决应用耦合,异步消息,流量削锋等问题。目前在生产环境,使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ等。由于每个消息队列都有它的优势和劣势,我们公司对于不同的场景使用了不同类型的消息队列。对于RocketMQ消费端存在消息消费失败的情况,通常有两种方式,一种是consumer端知道怎么处理,另一种是consumer不能处理(broker处理),本文对后一种情况进行介绍,consumer获取到消息但不能正常处理(ack),接下来这个消费失败的消息在Broker里面如何存储和重新让consumer消费,针对这个流程做了深入的分析。本文中的P代表producer,C代表consumer,本文的consumeQueue对应前面的topic下面的队列。

目录

  • RocketMQ的消费与存储结构
  • RocketMQ的消费失败消息处理逻辑
  • Broker端处理失败消息任务的启动
  • Consumer发回消费失败消息流程
  • Broker写发回失败消息的流程

RocketMQ的消费与存储结构

正常情况下,P发送消息到broker,消息内容写到commitlog,消息内容在commitlog的位置信息(索引)写到consumerQueue,C读取consumerQueue的内容消费消息。 流程图.png RocketMq的存储结构:
流程图.png 本文的内容涉及上面的消费队列服务(consumerQueue,%RETRY%groupName属于consumerQueue),定时消息服务(SCHEDULETOPICXXXX)两个模块,C与broker的的消息消费只涉及到consumerQueue,定时消息服务只在broker内部起作用。

RocketMQ的消费失败消息处理逻辑

consumer消费失败消息处理流程图如下:
流程图.png 在下面的代码和流程分析中请结合这个图进行分析。 其中SCHEDULETOPICXXXX和%RETRY%groupName的queue都存储在目录 ~/store/consumequeue 里面: ll ~/store/consumequeue 如下:
schedule and retry store.png ll ~/store/consumequeue/SCHEDULETOPICXXXX 如下:
schedule queue.png 从上图可以看出SCHEDULETOPICXXXX的队列名称是从2开始到17,对应的delayLevel为3到18,3对应10s,18对应2h,在类MessageStoreConfig中这样定义延时时间:String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"。SCHEDULETOPICXXXX这个topic只对内部使用,对于consumer只能消费到retry队列的数据。 consumer消费失败的消息发回broker后总是先写到SCHEDULETOPICXXXX里面,然后schedule service读取SCHEDULETOPICXXXX里面的数据写到retry队列,consumer消费retry队列的数据,这样就完成了一个循环,从这个过程也能看到,一个消费失败的消息体每次发回broker需要在commitLog里面存储两份(topic为SCHEDULETOPICXXXX的一份这个主要是为schedule service控制延时用的,topic为%RETRY%groupName的一份)。
当我们想查看现在的延时消息数量,我们可以查看SCHEDULETOPICXXXX的offset来得知,使用CLI Admin Tool工具输入命令“sh mqadmin brokerStatus”查看处理进度。如下图: shcedule-offset.png 其中每行为一个队列,图中第一列为队列的名称,图中第二列参数为当前队列处理的offset,图中第三列为当前队列最大存储的offset。通过第三列和第二列的值相减能得出当前的队列的消息数量。

Broker端处理失败消息任务的启动

Schedule Service Process Class Diagram0.png ScheduleMessageService根据messageDelayLevel维护了每个延迟level对应的队列编号,以及每个队列编号对应的offset。在start方法里面会启动18个timerTask(DeliverDelayedMessageTimerTask),每个对应一个level,初始offset为0。然后就是定时任务读取SCHEDULETOPICXXXX队列里面的消息进行判断,如果消息的delayLevel对应的时间满足重新消费,那么就会忘consumeQueue里面写这个消息,等待C重新来消费。

Consumer发回消费失败消息流程

Send Consume Failed Message Sequence Diagram1.png 在ConsumeRequest的run方法里面也就是业务端处理消息的线程里面,对于status是非success的交给ConsumeMessageConcurrentlyService(本文只讨论并行消费的模式,串行模式类似)的sendMessageBack方法处理,这个方法主要设置delayLevel(context.getDelayLevelWhenNextConsume()),然后传递给DefaultMQPushConsumerImpl.sendMessageBack找到对应的消息来源queue,把这个消息发送到这个queue里面,也就是说消费失败的消息发回broker还是会在之前的那个queue里面。发回broker后本地再过5秒重试消费一次,如果这次成功,下次就不再消费。 上面流程的类图: Send Message Class Diagram0.png 在ConsumeRequest的run方法里面会调用ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this)来处理消费结果状态,在cluster(集群模式)下设置新的消息delayLevle值然后把失败的消息发回Broker,广播模式不发回。注意ConsumeConcurrentlyContext的delayLevelWhenNextConsume属性说明-1直接放到死信队列,0又broker每次对重试消费次数加1来控制重试策略,大于0由consumer控制重试消费策略(在listener的consumeMessage方法里面有个context:context.setDelayLevelWhenNextConsume(4)设置为1分钟延时消费),默认值为0。

Broker写发回失败消息的流程

Schedule Sequence Diagram0.png broker端收到消费失败消息后通过consumerSendMsgBack(P发送的消息不由这个处理,区分通过消息头的type)方法设置当前消息的delayTimeLevel,这里计算delayTimeLevel,第一次重试默认consumer发回为0,延迟为延迟等级为0+3=3;如果第一次不为0表明是consumer控制的情况,直接取出delayTimeLevel,也就是和ConsumeConcurrentlyContext(consumer端控制)的delayLevelWhenNextConsume配置一致。设置好delayLevelTime后就交给DefaultMessageStore的putMessage方法,DefaultMessageStore的putMessage方法通过Commitlog的putMessage来写入文件,这里需要重点关注的是在这个方法里面通过msg.getDelayTimeLevel() > 0这个条件,修改当前消息topic为SCHEDULETOPICXXXX,原来的topic保留在property里面,在ScheduleMessageService里面判断消息满足条件后会把消息的topic改为真实的topic,通常是retry,接着写到consumeQueue里面,C对于%RETRY%consumerGroup这个topic在程序里面默认是订阅的不需用户指定,然后队列Id的计算方式为queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel()),即msg.getDelayTimeLevel()-1,和前面的截图2到17编号一致,然后消息体写到commitlog文件和索引写到SCHEDULETOPICXXXX队列。类图如下:
Process Reconsume Message Class Diagram0.png SendMessageProcessor处理远程发来的消息,包括P和C的,方法里面通过RequestCode.CONSUMERSENDMSG_BACK来判断是不是重试发回的消息。然后会判断这个消息对应的topic为%RETRY%consumerGroup的是否创建过,没有则创建;接下来的处理就和上面的流程图一样了。

总结

本文围绕consumer端消费失败后RocketMQ各个模块的处理逻辑进行了源码的深入分析。相信有了以上的知识学习和实践之后,当业务应用遇到了类似的问题就可以胸有成竹的应对了。