rocketmq消费问题总结

Posted 2023-03-09

参考技术A 流程就不写了，写点自己的总结，希望对各位有用，从解决问题的角度去观察RocketMq的设计思路，本人水平有限，说错的地方请及时指出。

问题一 RocketMq 消费流程

获取topic 对应的消费客户端和所有的broker 下的Que队列 然后根据一定的算法分配本客户端要拉取的QueID

分配算法

            1  环行平均分配算法，平均然后轮流分配

                q1 q2 q3 q4 q5 q6 q7 q8 3个消费队列 c1 c2 c3

                c1:q1 q4 q7

                c2:q2 q5 q8

                c3:q3 q6

            2 平均分配 

                q1 q2 q3 q4 q5 q6 q7 q8 3个消费队列 c1 c2 c3

                c1: q1 q2 q3

                c2: q4 q5 q6

                c3: q7 q8

            3机房优先平均分配

                优先分配同机房的消息队列，然后平均

           4自定义算法指定客户端消费某些队列算法。

            5 一致性hash算法

                算法原理 ：

                                步骤1 构造clientid的hash环,TreeMap 为集合，hash值作为key，节点为clientid

                                步骤2 计算que的hash 节点，获取> 本hash值得最近得一个clientid节点。

    1 多线程并行处理，不同得队列并行拉取数据，消息并行消费

默认一次拉取32消息，节省网络带宽

2 并发消费，不会因为中间得一个消息有问题，就停顿卡进度

    一个个消息消费，根据结果 反馈broker，有问题仍然 反馈给服务端进入重试队列，进行下一次得消费重试，不会因为有问题导致消费进度得卡顿

 

3 即使有一个消息有问题死循环，有超时检测机制

    对待处理得消息队列进行超时判断，超过了时间没处理完毕，发送回broker ，并从内存中移除。

答案

答案： 无论批次消费，还是一个个消费，坐标以小得为准。

目的： 防止后面得消费完，前面得消息还没消费完，服务宕机了，导致消息丢失。

源码：

        

存在得问题：为了确保消息不丢失，服务器重启得时候，会导致重复消费。

顺序消费的目的:客户端消费消息是按照，消息进入的顺序，并发消息，offset大的消息，有可能先于offset小的消息先行消费完。

1消费逻辑上，顺序消费，必须锁定broker 对应的消息队列，防止重新负载的时候分配给其他client

2顺序消费，一次拉取32条消息，如果中间有一条消息卡滞，消费失败，后面的消息挂起，这条消息重试16次，

如果还是失败，就会发回服务端。跳过此次消费。

问题七

顺序消费上了几把锁，为什么要上锁

1 负载均衡的时候，队列发生变化

目的：负载变化，要求broker给队列上锁，变更期间不允许分配给其他client

2数据拉取后，对队列数据的加锁，保持队列的顺序性消费

对集合上锁。

1 消息重试机制，消费失败的消息会重新发回broker端，

2 broker收到ack 响应才认为消费成功，否则不认为是成功

3客户端拉取一批消息，即使后面的先于前面的消费完，即使broker宕机，也只更新低的offset 确保消息不丢失

1消费失败会发回重试。根据重试的次数， 发往不同等级的重试队列

定时取出消息发往原来的topic 和que

达到最大失败次数放入死信队列

消息堆积有几种原因

消息堆积监控

1.判断是否存在消息堆积场景

       1producer发送消息的速率监控

       2producer发送消息的maxOffset与consumer消费消息的currOffset的差异值与给定的消息堆积数值告警值对比，如果差异   值大于数据告警值，则存在消息堆积，否则不存在消息堆积。

       3consumer消费消息的速率监控

通过扩容能解决问题的现象

        1 突然流量激增，导致堆积。

        2  Broker消息堆积，比如Broker的性能瓶颈，Broker同步策略导致消息堆积等

         3 consumer本身已经拉取消息的堆积。consumer消息拉取超过一定量之后会暂停消息拉取，一方面是消费者本身消费能力的现在，另一方面是由于消费端过多的消息容易造成GC频繁。

扩容还解决不了的问题，还存在挤压现象，就要考虑broker 或client本身的故障

        这种情况基本上是可以确定是RocketMQ本身的故障照成的，比如Broker故障，比如Broker的GC频率过高导致消息推送，copy性能降低，集群内部网络故障，等等。此时主要是监控RocketMQ服务器性能，或消费逻辑有问题

感谢以下作者辛苦的劳作参考

https://www.jianshu.com/p/45aa7465dfc1  ProcessQueue处理队列 作用

https://www.cnblogs.com/chenjunjie12321/p/7922362.html   消费端整体流程

https://my.oschina.net/javamaster/blog/1929879    流量控制

listener.consumeMessage 如果一直死循环不返回杂办？ Ack卡进度解决方案 http://jaskey.github.io/blog/2017/01/25/rocketmq-consume-offset-management/ rockmq 消费失败处理 rockmqack 机制具体解析
http://www.luyixian.cn/news_show_36799.aspx 并发消费和顺序消费区别
https://www.jianshu.com/p/fcfc662368f4 offset更新流程

并发消费任务后续任务是如何增加得

https://cloud.tencent.com/developer/article/1521811

 rebalance 解析

https://blog.csdn.net/yuxiuzhiai/article/details/103884106

消息堆积，解决方案 

https://blog.csdn.net/mlydaemon/article/details/105899869

https://blog.csdn.net/luanlouis/article/details/88078657

https://www.infoq.cn/article/NcSYj_2PQhBlqveuD1Kw 

offset游标更新方法

https://blog.csdn.net/fei33423/article/details/51255082

并发消费和顺序消费区别

https://www.cnblogs.com/allenwas3/p/12245217.html

RocketMq重试的场景

https://juejin.im/entry/5c9cf65ae51d453f77070d9b

rocketmq 框架需要解决得问题

https://www.cnblogs.com/yushangzuiyue/p/9684000.html

offset问题刨析

https://juejin.im/post/5d72724cf265da03be48fd24