RocketMQ生产者和消费者关键概念总结

墨蓝 2022-05-29 01:53 219阅读 0赞

**生产者**：

1.  每隔30S向Master发送心跳，心跳信息内只包含producerGroup名称，同时上传FilterClassSource
2.  每隔30S从Namesrv获取最新的TopicRouteData，提取信息封装成TopicPuhlishInfo
3.  根据最新的TopicRouteData，定时更新持有的Broker列表，清空下线的Broker
4.  当未开启延迟容错机制(默认)情形下，轮流按顺序向所有Broker的MessageQueue发送消息

**消费者**

1.  每隔30S向Broker所有节点发送心跳，发送groupName，消费模式（集群，广播），获取模式（push，pull），consumeFromWhere，SubscriptionData等信息
2.  每隔30S从Namesrv获取最新的TopicRouteData，提取所有的MessageQueue，存入RebalanceImpl的topicSubscribeInfoTable中，以备负载均衡时使用
3.  每隔30S根据获取到的最新TopicRouteData，更新Broker信息，移除下线的Broker
4.  每隔5S向Broker Master持久化进度，若Master宕机则向Slave持久化
5.  每隔20S进行一次负载均衡，随机从一个Broker上获取指定Group的所有Consumer的id集合(集群消费模式下)，将Client(消费者)和Master的MessageQueue排序，然后均分队列，不够的按ClientId序号分配余数。在Broker和Consumer不变的情况下，每个Client消费的MessageQueue不会变化，因为每次都是这些数据，排序后重新分配的结果和原来一样。当Broker和Consumer有变化后，才可能会改变原先消费的MessageQueue位置。
6.  RebalanceService获取到最新的Topic路由信息，当Broker或者Consumer有变化时，，更新Client持有的消费队列，将那些被不再属于自身的消费队列提交最新的消费进度，然后移除出消费请求列表，对于新增的消费队列，根据消费策略（consumeFromWhere）获取此队列相应的消费进度，生成消息拉取请求，马上执行消息拉取服务。
7.  在消费者启动时，会立即执行一次负载均衡，分配当前消费者订阅的MessageQueue，同时获取这些MessageQueue的相应消费进度，将这些信息封装成一个PullRequest，将其放入PullMessageService\#pullRequestQueue中，pullRequestQueue是一个阻塞队列LinkedBlockingQueue，当有PullRequest放入pullRequestQueue中，PullMessageService会立即从pullRequestQueue 弹出 PullRequest来执行。当第一次负载均衡结束后，put一个PullRequest进pullRequestQueue，PullMessageService就马上执行拉取信息请求了。
8.  **每一个消费队列的第一次拉取请求都是由RebalanceService在执行完负载均衡后产生**，之后的请求由上次拉取请求返回后生成，当负载均衡后若指定的消费队列不属于此Client，那么会设置其为丢弃的队列，此举会中断这种循环式的生成请求，也就是中断对不属于此Cilent的消费队列的请求。
9.  RocketMQ在拉取到消息后，会存入ProcessQueue的一个TreeMap类型的属性中，key为ConsumeQueue的Offset，value为MessageExt，消息在其中按Offset有序排列
10. 消费者在拉取到消息后，根据consumeMessageBatchMaxSize的大小指定每个线程消费消息的数量来拆分消息填入不同的消费者线程，然后提交到消费者线程池，执行消费
11. 消费者在消费完消息后(无论消息是否消费成功)，会根据消息的Offset，去TreeMap中移除相应的键值对。RocketMQ会每隔15分钟清除TreeMap中消费时间超过15分钟的消息(将消息发回Broker，同时从TreeMap中移除)，每15分钟最多移除16条过期消息
12. 并发式的消费者默认每次消费1条信息，当返回null，RECONSUME\_LATER，或者抛出异常时，默认消费失败，消费者会把当前消息发回给Broker，若发回失败，则在5S后重新消费此消息
13. 无论消息是否消费成功，均会更新到本地以及Broker上的进度表，因为消费失败的消息会被发回给Broker，存入%RETRY%+consumeGroup 的Topic中，消费者在启动时默认订阅重试Topic，发回给Broker的消息会根据delayLevel来判断重投递时间
14. 正常消费情形下更新进度只能增长，也就是多个线程并发更新进度时，只能将进度更新为当前之后，在下一次请求时会附带上当前消费的进度。
15. 在拉取消息时，如果是从Master处拉取，则在拉取消息时会附带更新Broker上的消费进度，消费者内有一个定时任务，每隔5S将消费进度持久化到Broker上，因为消费进度只会更得越来越大，所以两种更新形式执行时间晚的会覆盖之前的值。
16. 并发式的消息拉取，每次拉取的最大数量为32条消息，每个线程默认消费数量为1条，因此当一次拉去后，最多会生成32个消费者线程并发消费，消费者的消费线程池最小线程数为20，最大为64
17. 在拉取请求时要指定ConsumeQueue的起始位置，拉取的位置始终 >= 提交的消费进度位置 ，因为消费者可能还有线程在消费之前拉取的消息
18. 当拉取到消息后，将消息放进TreeMap中，然后提交消费请求(线程)到消费者线程池（如果消费者线程池已经满了，会阻塞放入消息，直到有空闲的线程），提交了之后就会马上发送下一次拉取请求，而不必等到拉取的消息消费完。如果Broker上没有新的消息，就会挂起此请求，直到生产者提交了新的消息，返回下一批消息
19. 基于以上原理，当Consumer突然宕机时，会丢失当前正在消费或者以消费完但是未提交的消费进度，Consumer恢复时会重复拉取这部分消息，所以要在消费者端做消息去重，或者保证消费的幂等性。