07 | 消息积压了该如何处理？

2019-08-06 李玥

《消息队列高手课》

课程介绍



讲述：李玥

时长:大小12.26M



你好，我是李玥。这节课我们来聊一聊关于消息积压的问题。
据我了解，在使用消息队列遇到的问题中，消息积压这个问题，应该是最常遇到的问题了，并且，这个问题还不太好解决。
我们都知道，消息积压的直接原因，一定是系统中的某个部分出现了性能问题，来不及处理上游发送的消息，才会导致消息积压。
所以，我们先来分析下，在使用消息队列时，如何来优化代码的性能，避免出现消息积压。然后再来看看，如果你的线上系统出现了消息积压，该如何进行紧急处理，最大程度地避免消息积压对业务的影响。
优化性能来避免消息积压在使用消息队列的系统中，对于性能的优化，主要体现在生产者和消费者这一收一发两部分的业务逻辑中。对于消息队列本身的性能，你作为使用者，不需要太关注。为什么这么说呢？
主要原因是，对于绝大多数使用消息队列的业务来说，消息队列本身的处理能力要远大于业务系统的处理能力。主流消息队列的单个节点，消息收发的性能可以达到每秒钟处理几万至几十万条消息的水平，还可以通过水平扩展 Broker 的实例数成倍地提升处理能力。
而一般的业...



展开全文

精选留言

约书亚
2019-08-06
批量消费有意义的场景要求：1.要么消费端对消息的处理支持批量处理，比如批量入库 2. 要么消费端支持多线程/协程并发处理，业务上也允许消息无序。3. 或者网络带宽在考虑因素内，需要减少消息的overhead。
批量消费的局限性：1. 需要一个整体ack的机制，一旦一条靠前的消息消费失败，可能会引起很多消息重试。2. 多线程下批量消费速度受限于最慢的那个线程。
但其实以上局限并没有影响主流MQ的实现了批量功能。
6
59
岁月安然
2019-08-06
1、要求消费端能够批量处理或者开启多线程进行单条处理
2、批量消费一旦某一条数据消费失败会导致整批数据重复消费
3、对实时性要求不能太高，批量消费需要Broker积累到一定消费数据才会发送到Consumer
作者回复: 👍👍👍
1
51
iLeGeND
2019-08-06
老师好，我一直理解，消息积压不是一种正常的现象吗？来不及处理的消息先在消息队列中存着，缓解下游系统的压力，让上下游系统在时间上解偶，，听了今天的课，感觉理解的不太一样，希望老师解答一下
作者回复: 消息积压是正常现象，积压越来越多就需要处理了。

就像一个水库，日常蓄水是正常的，但下游泄洪能力太差，导致水库水位一直不停的上涨，这个就不正常了。
5
47
C J J
2019-08-08
一、如何预防消息积压？
1、发送端提高并发及批量大小；
2、消费端增加实例且同步宽容分区；

二、如何处理消息积压？
1、消费端扩容；
2、服务降级；
3、异常监控。
展开
1
22
linqw
2019-08-06
尝试回答下课后习题，老师有空帮忙看下哦
消费端进行批量操作，感觉和上面的先将消息放在内存队列中，然后在并发消费消息，如果机器宕机，这些批量消息都会丢失，如果在数据库层面，批量操作在大事务，会导致锁的竞争，并且也会导致主备的不一致。如果是一些不重要的消息如对日志进行备份，就可以使用批量操作之类的提高消费性能，因为一些日志消息丢失也是可以接受的。
作者回复: 非常好！
1
18
Jxin
2019-08-06
1.无法提升消费业务效率（仅受消费业务自身逻辑影响），但可以提高mq中堆积消息消费的整体吞吐量（批推比单推mq耗时较短）。
2.数据增量同步，监控信息采集。（非核心业务的稳定大数据流操作）。
3.批处理意味数据积累和大数据传输，这会让单次消费的最长时延变长。同时批量操作为了保证当前批量操作一致性，在个别失败的情况下会引发批量操作重试。
作者回复: 总结的非常好！
16
严青
2019-12-02
我没读过这篇文章之前我看过别人面试的解决消息积压的方法：
（1）临时扩容，增加消费端，用硬件提升消费速度。
（2）服务降级，关闭一些非核心业务，减少消息生产。
（3）通过日志分析，监控等找到挤压原因，消息队列三部分，上游生产者是否异常生产大量数据，中游消息队列存储层是否出现问题，下游消费速度是否变慢，就能确定哪个环节出了问题
（4）根据排查解决异常部分。
（5）等待积压的消息被消费，恢复到正常状态，撤掉扩容服务器。
到此，问题解决
展开
14
lecy_L
2019-08-16
消息积压处理：
1、发送端优化，增加批量和线程并发两种方式处理
2、消费端优化，优化业务逻辑代码、水平扩容增加并发并同步扩容分区数量
查看消息积压的方法：
1、消息队列内置监控，查看发送端发送消息与消费端消费消息的速度变化
2、查看日志是否有大量的消费错误
3、打印堆栈信息，查看消费线程卡点信息
展开
作者回复: 👍👍👍
1
10
SunshineBoy
2020-03-04
如何判断增加多少consumer消费实例的个数？
作者回复: 你可以简单计算一下，消费并行度：单实例平均消费tps * 消费并行度 > 生产消息的总tps

消费并行度 = min（consumer实例数，分区数量）
8
天涯煮酒
2019-09-06
一开始不理解为啥Producer发得慢了会导致消息积压，明明是发得少了Broker端的消息应该更少啊

后来想到前面章节有提到消息事务，Producer首先会开启事务并发送一个半消息，再执行业务逻辑，最后提交事务，如果发得慢了会导致半消息在Broker中的时间增长，导致积压
2
6
长期规划
2019-09-02
老师，如果onMessage方法中，收到消息后不确认，等真正处理完消息再确认，就可以了吧，这样就可以用内存队列了
作者回复: 理论上是可以的，但你要注意，像RocketMQ，采用默认配置的时候，onMessage方法结束后，如果没抛异常，默认就会自动确认了。
1
6
亚洲舞王.尼古拉斯赵...
2019-08-06
如果使用了批量消费的方式，那么就需要批量确认，如果一次消费十条消息，除了第七条消费失败了，其他的都处理成功了，但是这中情况下broker只能将消费的游标修改成消息7，而之后的消息虽然处理成功了，但是也只能使用类似于拉回重传的方式再次消费，浪费性能，而且这种批量消费对于消费者的并发我觉得不是很友好，可能消费者1来了取走了十条消息在处理，这时候消费者2过来了也想取十条消息，但是他需要等待消费者1进行ack才可以取走消息，不知道说的对不对，请老师指正
作者回复: 是这样的。
7
grey927
2019-08-26
发送端如果多线程，如何实现之前您在《确保消息不会丢失》这篇文章中说的：
发送端把发送消息标识为每次增加1的效果？
作者回复: 可以使用一个自增的原子变量，比如Java中的AtomicLong。
1
5
Stenvien
2019-08-11
如果消费者消费异常，即使多次消费也无法成功处理（如消息格式异常），导致一直无法成功ack此条消息，这种场景一般要怎么处理？

我想到有2种：
1. 不做任何处理，消费者会一直卡在此消息的处理上，那么后面的所有消息都没机会处理了，只能靠监控发现消费延迟，发出告警，人工修复。这种处理方式会导致一条有问题的消息就影响了整个业务
2. 数据库存储此异常的消息，并发告警，人工修复，仍然ack此消息，继续消费后面的消息。但是，若对消息的处理顺序有依赖，若没有成功处理此异常消息，消费的后面的消息的处理可能会有问题

是否有更好的处理方式？
展开
作者回复: 有的消息中间件提供了“死信队列”的功能，它会自动把这种反复消费都失败的消息丢到一个特殊的死信队列中，避免一条消息卡主队列的情况。
5
涛涛
2020-04-13
临时扩容消息分区，已堆积的消息会转移到新分区上吗？
作者回复: 不会的。
4
传志
2020-03-09
老师想问下，为什么生产端性能问题怎么会引起消息堆积呀
作者回复: 生产端发送慢不会引起消息积压的。
3
13761642169
2019-08-09
消息本来就有限流或者削峰填谷，这个也是使用消息队列的作用，老师您讲的不严谨。
1
3
蓝魔丶
2019-08-06
消费端通过丢入队列，并使用多线程来提高并发，老师讲到可能丢消息，但也要是某种条件成立，比如开启了自动提交，如果改为手动提交完全可以避免丢消息，只是可能出现重复消费问题，至少不会丢
5
3
@%初%@
2020-12-16
消息堆积，通过扩容broker的分区数，这点我有不同的见解，增加分区数量，势必会引发consumer端rebalance发生，如果消息不是分区内顺序性的，没有问题，但是如果是顺序消息，那么rebalance之后，消息的顺序性难以保证，这样对于消费端，可能就会出现问题，针对此方式，在重写分区策论时一定要注意，是否可以满足扩容的需求，不然就会出现，消息堆积问题解决了，但是出现业务错误，有点得不偿失。
2
Get it
2020-04-15
老师，你好
想请问下在消费端一般通过先处理业务逻辑再确认消息以保证消息不丢失
但是如果业务逻辑有未知风险会持续抛出异常导致消息一直无法消费导致积压一般如何解决呢
作者回复: 有些消息队列有针对这个问题的解决方案，比如RocketMQ有死信队列，对于多次消费的失败的消息，扔到死信队列中，避免坏消息卡住队列。
2