rocketmq消费问题总结
流程就不写了,写点自己的总结,希望对各位有用,从解决问题的角度去观察RocketMq的设计思路,本人水平有限,说错的地方请及时指出。
问题一 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节点。
问题三 拉取offset下标如何计算
问题四 rocketmq consume客户端 高并发体现在哪?
1 多线程并行处理,不同得队列并行拉取数据,消息并行消费
默认一次拉取32消息,节省网络带宽
2 并发消费,不会因为中间得一个消息有问题,就停顿卡进度
一个个消息消费,根据结果 反馈broker,有问题仍然 反馈给服务端进入重试队列,进行下一次得消费重试,不会因为有问题导致消费进度得卡顿
3 即使有一个消息有问题死循环,有超时检测机制
对待处理得消息队列进行超时判断,超过了时间没处理完毕,发送回broker ,并从内存中移除。
问题五
rocketmq 消费offset 反馈机制是如何更新得?批量消费中,如下图所示,消费完是更新坐标最小得,还是更新坐标做大得,为什么?多线程,单个消费中,后面得先于前面得消费完,坐标是更新得是小得,还是大得。?
答案
答案:无论批次消费,还是一个个消费,坐标以小得为准。
目的:防止后面得消费完,前面得消息还没消费完,服务宕机了,导致消息丢失。
源码:
存在得问题:为了确保消息不丢失,服务器重启得时候,会导致重复消费。
问题六 顺序消费 流程,顺序消费和并发消费有什么不同。
顺序消费的目的:客户端消费消息是按照,消息进入的顺序,并发消息,offset大的消息,有可能先于offset小的消息先行消费完。
1消费逻辑上,顺序消费,必须锁定broker 对应的消息队列,防止重新负载的时候分配给其他client
2顺序消费,一次拉取32条消息,如果中间有一条消息卡滞,消费失败,后面的消息挂起,这条消息重试16次,
如果还是失败,就会发回服务端。跳过此次消费。
问题七
顺序消费上了几把锁,为什么要上锁
1 负载均衡的时候,队列发生变化
目的:负载变化,要求broker给队列上锁,变更期间不允许分配给其他client
2数据拉取后,对队列数据的加锁,保持队列的顺序性消费
对集合上锁。
问题八rocketmq如何确保消息不丢失?
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