RocketMQ Consumer消息消费过程消息的处理

Posted 2022-02-10 乐观男孩

目录

• 说明
• 消息处理流程图
• 拉取结果处理
• ConsumeMessageOrderlyService
• 总结

说明

从上一节《Consumer消息消费过程（二）、消息的拉取》中，我们知道，拉取到消息后，会通过调用回调函数的方式对消息进行处理。在本节中，我们就学习下消息的处理过程。

消息处理流程图

拉取结果处理

1、如果消息拉取失败，则直接调用回调函数的onException方法，该方法会重新将消息拉取请求放入PullMessageService的阻塞队列，重新放入的任务会间隔一段时间（默认3秒）后才执行

2、如果拉取到消息，则调用回调函数的onSuccess()方法，进行如下处理：
（1）、调用PullAPIWrapper.processPullResult()方法进行结果处理：过滤消息（从这里可见，是拉取到消息再进行过滤）、设置对应的属性


（2）、如果拉取到消息：
a、设置下一次消费的offset到PullRequest中、统计相应的信息

b、将当前拉取到的消息设置到processQueue、提交消息到ConsumeMessageService（ConsumeMessageService分ConsumeMessageConcurrentlyService和ConsumeMessageOrderlyService，具体由Consumer注册的监听器决定。这里以ConsumeMessageConcurrentlyService为例子讲解）

ConsumeMessageConcurrentlyService处理消息：这里可以看到，拉取到的消息，会分批封装成ConsumeRequest，然后提交给线程池（这里是一个异步处理，而且是多个线程并发处理）进行处理。

ConsumeRequest的run方法，其实就是调用我们为Consumer注册的监听器（defaultMQPushConsumer.registerMessageListener），执行具体的业务逻辑处理

对执行监听器的结果进行处理

这里需要注意，因为调用监听器就是执行我们具体的业务逻辑，业务逻辑有可能会发生异常，处理消息失败（比如发生异常或者返回false），如果处理消息失败，RocketMQ会隔一段时间后（默认5秒）再重新将消息提交给ConsumeMessageConcurrentlyService的线程池，后续再重新将消息推送给监听器。


可见，如果监听器执行业务处理失败，消息是缓存在内存中，并且线程池中的线程数也有可能会耗尽。
c、进行下一次的消息拉取

ConsumeMessageOrderlyService

ConsumeMessageService有ConsumeMessageConcurrentlyService和ConsumeMessageOrderlyService两个实现类。ConsumeMessageConcurrentlyService是多线程并发对消息进行处理，不保证按消息的顺序进行处理。而ConsumeMessageOrderlyService保证按照消息的顺序进行处理。下面，我们看一下ConsumeMessageOrderlyService如何保证按消息的顺序进行处理。
进入ConsumeMessageOrderlyService.submitConsumeRequest，可以看到，ConsumeMessageOrderlyService对拉取到的消息，整批提交个线程池处理（ConsumeMessageConcurrentlyService会分批次提交，所以达到并发的效果）：

ConsumeRequest.run()方法，会对整个消息队列MessageQueue进行加锁，从而保证同一时刻只有一个线程对MessageQueue进行处理：

然后从ProcessQueue（MessageQueue和ProcessQueue是一对一的关系，所以从Broker对应的queue拉取到的消息，都同一放在了ProcessQueue里）中取出消息进行处理：

takeMessages()是按顺序将消息取出：

取出之后，交给消息监听器处理，进行业务处理，然后返回处理结果：

对返回的结果进行处理

如果处理成功，则继续处理其它消息（返回true）；如果处理失败（返回false，当前线程不再处理后面的消息，等后面重试再同一处理，所以当前线程就算任务结束了），则重新将当前的消息放回ProcessQueue中，并稍后再提交任务到线程池。


从以上的处理步骤，我们知道，有几个关键点：
1、从Broker对应的queue拉取下来的消息，统一存放在ProcessQueue的treeMap中，该map已经对消息的offset进行排序。
2、虽然每拉取到一批消息，就提交到ConsumeMessageOrderlyService进行处理，线程池中会有多个线程对消息进行处理，但是由于对整个MessageQueue进行了加锁操作，所以只有一个线程能拿到锁，拿到锁的线程有可能对所有消息进行处理，后续的线程即使拿到锁，也无消息可以处理，直接完成。
3、当拿到锁的线程从ProcessQueue拿出消息时，是按照offset顺序拿的，从而保证按顺序处理消息。
4、如果业务处理失败，会把当前处理的消息重新放回ProcessQueue中，当前线程直接结束，后续重新启动线程对消息进行处理。所以即使业务处理失败，也能保证按消息的顺序进行处理。

总结

1、消息拉取和消息处理两个处理是分开的，拉取到消息，会提交给对应ConsumeMessageService进行处理。
2、消息是由ConsumeMessageService异步推送给Consumer的监听器处理。
3、如果监听器处理消息失败，消息将缓存在内存中，有可能占用过多的内容。
4、如果监听器处理消息失败，还可能导致ConsumeMessageService线程池中的线程（默认最小和最大线程数都是20）耗尽。该线程池的线程数可以在定义Consumer中定义。