consumer的每个实例是靠队列分配来决定如何消费消息的。那么消费进度具体是如何管理的,又是如何保证消息成功消费的?(RocketMQ有保证消息肯定消费成功的特性,失败则重试)什么是ACK
消息确认机制在实际使用RocketMQ的时候我们并不能保证每次发送的消息都刚好能被消费者一次性正常消费成功,可能会存在需要多次消费才能成功或者一直消费失败的情况,那作为发送者该做如何处理呢?
PushConsumer为了保证消息肯定消费成功,只有使用方明确表示消费成功,RocketMQ才会认为消息消费成功。中途断电,抛出异常等都不会认为成功——即都会重新投递。代码示例
消费的时候,我们需要注入一个消费回调,具体sample代码如下:
RocketMQ是以consumer group+queue为单位是管理消费进度的,以一个consumer offset标记这个这个消费组在这条queue上的消费进度。如果某已存在的消费组出现了新消费实例的时候,依靠这个组的消费进度,就可以判断第一次是从哪里开始拉取的,每次消息成功后,本地的消费进度会被更新,然后由定时器定时同步到broker,以此持久化消费进度。
但是每次记录消费进度的时候,只会把一批消息中最小的offset值为消费进度值,如下图:
部分源码如下:
核心源码如下:
通过这个逻辑对比我定制的时间,可以看出有几个不太完善的问题:
对于无序消息(普通、定时、延时、事务消息),当消费者消费消息失败时,您可以通过设置返回状态达到消息重试的结果。无序消息的重试只针对集群消费方式生效;广播方式不提供失败重试特性,即消费失败后,失败消息不再重试,继续消费新的消息。
注意 以下内容都只针对无序消息生效。重试次数
消息队列RocketMQ版默认允许每条消息最多重试16次,每次重试的间隔时间如下。第几次重试与上次重试的间隔时间第几次重试与上次重试的间隔时间110秒97分钟230秒108分钟31分钟119分钟42分钟1210分钟53分钟1320分钟64分钟1430分钟75分钟151小时86分钟162小时如果消息重试16次后仍然失败,消息将不再投递。如果严格按照上述重试时间间隔计算,某条消息在一直消费失败的前提下,将会在接下来的4小时46分钟之内进行16次重试,超过这个时间范围消息将不再重试投递。
注意 一条消息无论重试多少次,这些重试消息的Message ID不会改变。和生产端重试区别
消费者和生产者的重试还是有区别的,主要有两点
消费失败后,重试配置方式,集群消费方式下,消息消费失败后期望消息重试,需要在消息监听器接口的实现中明确进行配置(三种方式任选一种):
集群消费方式下,消息失败后期望消息不重试,需要捕获消费逻辑中可能抛出的异常,最终返回Action.CommitMessage,此后这条消息将不会再重试。示例代码
消费者收到消息后,可按照以下方式获取消息的重试次数:
消费端只发送一条消息进行测试重试消费端主动拒绝
消费端返回 ConsumeConcurrentlyStatus.RECONSUME_LATER,消费端就会不断的重试。
这里的代码意思很明显: 主动抛出一个异常,然后如果超过3次,那么就不继续重试下去,而是将该条记录保存到数据库由人工来兜底。
这里的超时异常并非真正意义上的超时,它指的是指获取消息后,因为某种原因没有给RocketMQ返回消费的状态,即没有return ConsumeConcurrentlyStatus.CONSUME_SUCCESS 或 return ConsumeConcurrentlyStatus.RECONSUME_LATER那么 RocketMQ会认为该消息没有发送,会一直发送。因为它会认为该消息根本就没有发送给消费者,所以肯定没消费。
当获得 当前消费重试次数为 = 0 后 , 关掉该进程。再重新启动该进程,那么依然能够获取该条消息
重启消费者
一般情况下我们在实际生产中是不需要重试16次,这样既浪费时间又浪费性能,理论上当尝试重复次数达到我们想要的结果时如果还是消费失败,那么我们需要将对应的消息进行记录,并且结束重复尝试。
在介绍RocketMQ的消费重试机制之前,需要先来说下“重试队列”和“死信队列”两个概念。重试队列
在业务工程中的Consumer端(Push消费模式下),如果消息能够正常消费需要在注册的消息监听回调方法中返回CONSUME_SUCCESS的消费状态,否则因为各类异常消费失败则返回RECONSUME_LATER的消费状态。消费状态的枚举类型如下所示:
业务方正常消费正常情况下,设置ackIndex的值为consumeRequest.getMsgs().size() - 1,因此后面的遍历consumeRequest.getMsgs()消息集合条件不成立,不会调用回发消费失败消息至Broker端的方法—sendMessageBack(msg, context)。最后,更新消费的偏移量;
业务方消费失败异常情况下,设置ackIndex的值为-1,这时就会进入到遍历consumeRequest.getMsgs()消息集合的for循环中,执行回发消息的方法—sendMessageBack(msg, context)。这里,首先会根据brokerName得到Broker端的地址信息,然后通过网络通信的Remoting模块发送RPC请求到指定的Broker上,如果上述过程失败,则创建一条新的消息重新发送给Broker,此时新消息的Topic为“%RETRY%+ConsumeGroupName”—重试队列的主题。其中,在MQClientAPIImpl实例的consumerSendMessageBack()方法中封装了ConsumerSendMsgBackRequestHeader的请求体,随后完成回发消费失败消息的RPC通信请求(业务请求码为:CONSUMER_SEND_MSG_BACK)。倘若上面的回发消息流程失败,则会延迟5S后重新在Consumer端进行重新消费。与正常消费的情况一样,在最后更新消费的偏移量;
疑问:上面说了RocketMQ的重试队列的Topic是“%RETRY%+consumerGroup”,为啥这里要保存至Topic是“SCHEDULE_TOPIC_XXXX”的这个延迟队列中呢?在源码中搜索下关键字—“SCHEDULE_TOPIC_XXXX”,会发现Broker端还存在着一个后台服务线程—ScheduleMessageService(通过消息存储服务—DefaultMessageStore启动),通过查看源码可以知道其中有一个DeliverDelayedMessageTimerTask定时任务线程会根据Topic(“SCHEDULE_TOPIC_XXXX”)与QueueId,先查到逻辑消费队列ConsumeQueue,然后根据偏移量,找到ConsumeQueue中的内存映射对象,从commitlog日志中找到消息对象MessageExt,并做一个消息体的转换(messageTimeup()方法,由定时延迟队列消息转化为重试队列的消息),再次做持久化落盘,这时候才会真正的保存至重试队列中。看到这里就可以解释上面的疑问了,定时延迟队列只是为了用于暂存的,然后延迟一段时间再将消息移入至重试队列中。RocketMQ设定不同的延时级别delayLevel,并且与定时延迟队列相对应,具体源码如下:
每个Consumer实例在启动的时候就默认订阅了该消费组的重试队列主题,DefaultMQPushConsumerImpl的copySubscription()方法中的相关代码如下:
最后,给出一张RocketMQ消息重试机制的框图(ps:这里只是描述了消息消费失败后重试拉取的部分重要过程):
本文由传智教育博学谷狂野架构师教研团队发布。
如果本文对您有帮助,欢迎关注和点赞;如果您有任何建议也可留言评论或私信,您的支持是我坚持创作的动力。
转载请注明出处!
欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/) | Powered by Discuz! X3.4 |