RocketMQ观后感--Producer，Consumer

系统管理员 2022-05-27 02:08 200阅读 0赞

![这里写图片描述][SouthEast]

先讲Producer的总结，之后讲Consumer的总结。

Producer观后感总结：

1.  Producer与NameServer维护长连接，每隔30S获取其发送消息topic的最新数据，覆盖上次的MessageQueue。
2.  Producer发送消息时，默认按顺序依次选择MessageQueue发送，当某个Broker宕机时，其MessageQueue发送不成功，那么会按顺序找到下一个不是此Broker的MessageQueue发送。
3.  若想要实现**全局顺序消费**，那么顺序发送是其前提。选择同步发送形式，借助MessageQueueSelector将此topic的所有消息发送到同一个MessageQueue下，且只能有一个Producer。
4.  若想要实现**局部顺序消费**，比如一个业务的多条信息有序消费，那么选择同步发送形式，借助MessageQueueSelector将当前业务的多条消息发送到同一个MessageQueue。
5.  Producer每隔30S向其订阅topic所在的所有Broker发送心跳信息，心跳信息仅包含ProducerGroup。Broker为与Producer的长连接注册了ChannelEventListener，当Producer宕机时，立即删除此Producer的心跳信息。
6.  Producer可发送事务消息，使用TCC模式，Confirm/Cancle时是Oneway形式，可能会发送失败。RocketMQ Version 3 \_1 \_1之后已经不支持事务消息回查了，因为原有的事务消息模式时在Confirm/Cancle时修改Half消息的状态，这会导致内存中脏页太多，有隐患。所以发送事务消息时需要在自己做好回查工作，可以借鉴我的一篇博客，[RocketMQ事务消息回查设计方案][RocketMQ]。
7.  一个Producer可以发送多个topic的消息，同步模式下一条发完有返回结果后再发下一条；异步模式下一条发完不等待结果马上发下一条，有结果后执行SendCallback。也就是说发送过程是同步的。
8.  Producer在发送消息时，会首先获取TopicRouteData。如果是新的消息，由于Broker上面还没有创建对应的Topic，没有有效数据返回，此时Producer尝试获取默认TOPIC的（RocketMQ会在每台broker上面创建名为TBW102的TOPIC）TopicRouteData，然后使用TopicRouteData作为模板发送消息。Producer会随机选择一台Broker发送消息，选中的broker在存储消息时，发现消息的topic还没有创建，若Broker的配置autoCreateTopicEnable==true(默认值)，就会按照TBW102的模板创建topic相关信息，生成ConsumeQueue。然后Broker将此topic的数据注册到NameServer，下一次发送此消息时Producer就能正常的从NameServer获取到TopicDataRoute了。 后果：以后此topic的消息都只能发送到此Broker上，失去了负载均衡的能力了。建议线上环境Broker的autoCreateTopicEnable=false，同时手动创建topic：

public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
        createTopic(key, newTopic, queueNum, 0);
    }

key使用TBW102，这个key的作用就是从NameServer处获取所有含有newTopic的Broker的brokerAddr，每个Broker都默认含有此TBW102，所以此方式能向所有的Broker创建指定的topic。

所以基于目前RocketMQ的设计，建议关闭自动创建TOPIC的功能，然后根据消息量的大小，手动创建TOPIC。且新增的Broker只有手动创建topic后，才能加入消息的存储和消费体系中。

Consumer观后感总结(基于Push形式)：

1.  Consumer按照消息拉取模式分为Pull和Push两种，他们的实质都是客户端主动拉取，只不过Push模式RocketMQ框架帮我们做了很多拉取工作。而Pull模式需要我们手动执行pullMessage方法，按轮询形式对每个MessageQueue拉取消息。如果是并发消费形式的话还需要按每个线程消费量大小分线程，然后多线程消费。如果MQ水平扩容，Pull形式不能被动的执行负载均衡，只能定期的去NameServer更新MessageQueue。很麻烦，所以一般建议使用Push形式的消费者。
2.  Consumer在start时，会先从NameServer处获取其订阅的topic的TopicRouteData，然后生成MessageQueue。
3.  然后向此含有此topic的所有Broker发送心跳信息，将自身的信息注册到Broker生成ConsumerGroupInfo，其包含了consumeGroup，Channel，topic等等信息。
4.  最后执行负载均衡。首先从随机从某个Broker获取此topic的所有Consumer的clientId，clientId是由Consumer在初始化时生成，通过心跳信息注册到Broker的，有几个clientId就代表此topic有几个Consumer。对所有的clientId和MessageQueue进行排序，然后按照分配策略(默认均分)来提取自身clientId分配到的MessageQueue，然后就发起对自身MessageQueue的消息拉取请求。
5.  Broker为所有Client(Producer，Consumer)的长连接都注册了统一的ChannelEventListener，当某个Client的长连接中断时，触发监听事件。首先遍历ProducerManager里所有Producer的心跳信息，通过Channel进行匹配，如果匹配上，说明断开的是Producer，那么移除此Producer相关的注册信息；如果匹配不上，那么不对ProducerManager里信息做处理。
6.  其次遍历ConsumerManager里所有的ConsumerGroupInfo，通过Channel去匹配其是否属于某个ConsumerGroupInfo，如果匹配上，那么说明断开的是一个Consumer。移除ConsumerGroupInfo里的断开的Channel和clientId，移除后如果ConsumerGroupInfo里已经不存在Channel了，说明这个消费者组已经没有消费者了，那么移除ConsumerGroupInfo。如果还有Channel，那么通知剩余的每个Channel对应的Client，让其立即执行负载均衡。
7.  仅有Consumer宕机，MessageQueue不变，得到通知的Consumer立即向Broker请求其订阅topic的所有clientId。然后按照均分策略重新分配MessageQueue，如果clientId有变化，那么重新分配到的MessageQueue和之前的肯定不一样，此时就需要移除不再属于自己的MessageQueue，同时向新分配的MessageQueue发起消息拉取请求。
8.  新增Consumer时，其向所有相关的Broker注册心跳信息，ConsumerGroupInfo新增了clientId，但ChannelEventListener里没有对新增连接事件做处理，也就是其他消费者在下一次负载均衡时才能更新MessageQueue。而新增的消费者能通过负载均衡马上分配到MessageQueue。在并发模式下，会造成一个MessageQueue两个消费者在消费，所以客户端要去重。以上就是负载均衡的实现流程。
9.  Push模式消费按消息处理形式分为有序式和并发式两种。
10. 并发模式消费者，可指定每次拉取消息的最大条数，默认值是32，也就是每次最多拉取32条消息；可指定每个消费线程消费数量，默认值是1，也就是MessageListenerConcurrently里的消费方法，List< MessageExt> 里面只有一条消息；并发式消费线程池的上限是64；拉取消息是单线程的，也就是拉取到一批消息后，按每个线程的消费容量拆分消费线程，提交进线程池后，不等待消费结果，马上执行下一次消息拉取；也就是单线程拉取，多线程消费。
11. 有序模式消费者，通过MessageQueue的分布式锁，ProcessQueue消费锁，ProcessQueue内部消息锁来**确保同一时间一个MessageQueue只能被一个Consumer内的一个线程消费**。因此当Producer有序发送时，Consumer就能实现严格有序消费。
12. Consumer可以配置消息的最大消费次数，默认值是16。当消息消费失败时，会sendMessageBack 回Broker，Broker通过ScheduleMessageService实现对发回消息的延时投递，延时级别为：consumeTime+3，延时级别与时间的配置为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，也就是第一次发回后，0+3 >> 10S 后重新投递。**每发回一次就代表着此消息在CommitLog重新存储两次**。存储第一次时生成的PositionInfo存入SCHEDULE\_ TOPIC\_XXXX中，tagsCode存储该被投递的时间戳。然后ScheduleMessageService会在tagsCode代表的时间解析此PositionInfo，从CommitLog提取消息体，再次存入CommitLog。这次生成的PositionInfo存入 %RETRY%+consumeGroup中。每个消费者启动时都默认订阅%RETRY%+consumeGroup这个topic，当此ConsumeQueue有消息时，马上拉取消费。
13. 当消费次数超过maxReconsumeTimes时，当前消息将不会再投递，而会被存入死信队列，topic：%DLQ%+consumeGroup，死信队列只能写入不能读取，其在创建时就设置此权限。Consumer能从NameServer处获取死信队列的TopicRouteData，但其生成MessageQueue时，有读取权限判断，死信队列不能读取，其会生成ConsumeQueue，但是不会生成MessageQueue，也就是不能被消费。所以在处理业务时，消费失败时对这种情况做相应处理。
14. 可配置消息的最大消费次数，默认值16太多了，可以自定义maxReconsumeTimes 的值，比如3次。
15. Client(Producer，Consumer)有很多好用的API，比如viewMessage，queryMessage，searchOffset等等，他们有助于加强对RocketMQ的使用理解。

[SouthEast]: /images/20220527/b71466fdefb14d739c479dbf215abb5a.png
[RocketMQ]: https://blog.csdn.net/qq_27529917/article/details/79802406