春招面试复习：消息队列（五）- 消息积压解决方案

港控/mmm° 2023-01-06 11:35 143阅读 0赞

# 1 消息积压问题追溯 #

系统某部分出现性能问题，来不及处理上游发的消息，导致消息积压。  
消息积压是正常现象，积压越来越多就需要处理了。  
就像一个水库，日常蓄水是正常的，但下游泄洪能力太差，导致水库水位一直不停的上涨，这个就不正常了。

# 2 积压的开发警示 #

防范于未然，平时使用MQ时，如何优化代码才能避免消息积压？  
若线上已出现积压，如何处理最好？

# 3 性能优化 #

性能的优化主要在生产者和消费者这俩业务逻辑。

MQ自身性能，作为API使用者，无需过于关注。  
因大多MQ业务，MQ本身处理能力远大于业务系统。主流MQ的单个节点，消息收发性能可达几w ~ 几十w条消息/s，还可水平扩展Broker实例数倍增处理能力。  
而一般业务系统需处理的业务逻辑远比消息队列复杂，单节点每秒可处理几百~几千次请求，已算性能佳。  
所以，MQ性能优化，更关注在消息收发两端，业务代码怎么和MQ协作达到最佳性能。

## 3.1 生产端 ##

此端业务代码处理性能，实际上和MQ关系不大，因都是先执行自己的业务逻辑，最后再发送消息。  
若你的代码发送消息的性能上不去，需优先检查，是否为发消息前的业务逻辑耗时太多导致。

对于发消息的业务逻辑，只需注意设置合适并发和同步大小，即可达到很好发送性能。为何？  
Producer发消息给Broker，Broker收到消息后返回确认响应，是一次完整交互。假设一次交互平均时延1ms，把这1ms分解：

1.  发送端准备数据、序列化消息、构造请求等逻辑时间，即发送端在发送网络请求前的耗时
2.  发送消息和返回响应在网络传输中耗时
3.  Broker处理消息的时延

若单线程发送，每次只发1条，则每秒只发

1000ms / 1ms * 1条/ms = 1000条消息

这并不能压榨MQ性能。

无论是增加每次发送消息的批量大小、增加并发，都能倍增发送性能。  
那么选择批量发送还是增加并发呢？  
还是那句话，取决于发送端的业务性质。能满足你的性能要求，就能搞。

*  若发送端是个微服务，主要接受RPC请求处理在线业务。  
    微服务在处理每次请求时，就在当前线程直接发消息即可，因为所有RPC框架都是多线程支持并发，自然可并行发送消息。  
    且在线业务比较在意请求响应时延，批量发送势必影响RPC服务时延。  
    这时通过并发提升发送性能就更好。
 *  若是离线分析系统，并不关心时延，而注重整个系统的吞吐量。发送端数据都来自DB，这就更适合批量发送，可批量从DB读数据，然后批量来发送消息，同样用少量并发即可获得高吞吐量。

> 批量消费中如果某条消息消费失败那么重试是会将整批消息进行重发。  
> 批量消费是一次取一批消息，等这一批消息都成功了，再提交最后一条消息的位置作为新的消费位置。如果其中任何一条失败，则认为整批都失败。

### 3.2 消费端 ###

使用MQ候，大部分性能问题都出现在消费端。若消费速度跟不上发送端生产消息速度，就会造成消息积压。若这种性能倒挂的问题只是暂时的，那问题不大，只要消费端的性能恢复之后，超过发送端的性能，那积压的消息是可以逐渐被消化掉的。

若消费速度一直比生产速度慢，久而久之，系统就会异常。

*  要么，MQ存储被填满无法提供服务
 *  要么消息丢失

所以设计系统，要保证`消费端消费性能高于生产端发送性能`，系统才能长治久安。

消费端性能优化除优化消费业务逻辑，也可水平扩容，增加消费端并发数提升总体消费性能。  
扩容Con实例数量时，必须同步扩容主题中的分区（也叫队列）数量，确保Con实例数和分区数量相等。  
若Con实例数量>分区数量，这样的扩容实际上徒劳。  
因为对消费者，在每个分区实际上只能支持单线程消费。

> 这一步需要业务consumer团队联系消息中间件团队一起运维配合。

我见到过很多消费程序，他们是这样来解决消费慢的问题的：  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTg5NTEw_size_1_color_FFFFFF_t_70][]

它收消息处理的业务逻辑可能较慢，也很难再优化，为避免积压，在收消息的OnMessage方法，不处理任何业务，把这消息放到一个内存队列就返回了。然后可启动很多业务线程，里面是真正处理消息的业务逻辑，这些线程从内存队列取消息处理，就解决了单Consumer不能并行消费问题。

这完美地实现了并发消费？  
这是一个常见错误方法！ 会丢消息。  
如果收消息的节点宕机，内存队列中还没处理的消息就会丢失。

# 消息积压，如何处理 #

还有种消息积压，日常系统正常运转时，没有积压或只有少量积压很快就消费了，但某个时刻，突然开始积压消息且积压持续上涨。  
这种情况下需要你在短时间内找到消息积压原因，迅速解决问题才不至于影响业务。

突然积压原因多种多样，不能一概而论。但排查消息积压原因，有相对固定且有效方法。

积压突然增加，最粗粒度的原因，只有两种

*  要么发送变快
 *  要么是消费变慢

大部分消息队列都内置了监控的功能，只要通过监控数据，很容易确定是哪种原因。如果是单位时间发送消息增多，比如说是赶上抢购，短时间内不太可能优化消费端代码提升性能，唯一方法通过扩容消费端实例数提升总体消费能力。  
如果短时间没足够服务器资源扩容，将系统降级，通过关闭一些不重要业务，减少发送方发送数据量，最低限度让系统还能正常运转，服务一些重要业务。

> kafka不仅是扩容时候，只要是consumer和partition有一方的数量变化，都会触发reblance。

还有种不太常见的，通过监控发现，无论是发送消息速度还是消费消息速度和原来都没什么变化，这时需检查消费端，是不是消费失败导致的一条消息反复消费，也会拖慢整个系统消费速度。

如果监控到消费变慢，你需要检查消费实例，分析一下是什么原因导致消费变慢。优先检查一下日志是否有大量消费错误，如果没有错误，可通过打印堆栈信息，看消费线程是不是卡在什么地方不动，比如**触发死锁或卡在等待某些资源**。

> 如果消费者消费异常，即使多次消费也无法成功处理（如消息格式异常），导致一直无法成功ack此条消息，这种场景一般要怎么处理？  
> 有的MQ提供了“死信队列”的功能，它会自动把这种反复消费都失败的消息丢到一个特殊的死信队列中，避免一条消息卡主队列情况。

# 总结 #

消息积压处理：  
1、发送端优化，增加批量和线程并发两种方式处理  
2、消费端优化，优化业务逻辑代码、水平扩容增加并发并同步扩容分区数量  
查看消息积压的方法：  
1、消息队列内置监控，查看发送端发送消息与消费端消费消息的速度变化  
2、查看日志是否有大量的消费错误  
3、打印堆栈信息，查看消费线程卡点信息

1.无法提升消费业务效率（仅受消费业务自身逻辑影响），但可以提高mq中堆积消息消费的整体吞吐量（批推比单推mq耗时较短）。  
2.数据增量同步，监控信息采集。（非核心业务的稳定大数据流操作）。  
3.批处理意味数据积累和大数据传输，这会让单次消费的最长时延变长。同时批量操作为了保证当前批量操作一致性，在个别失败的情况下会引发批量操作重试。

*  消费端是否可以通过同步消费提升消费性能呢？  
    消费端进行批量操作，感觉和上面的先将消息放在内存队列中，然后在并发消费消息，如果机器宕机，这些批量消息都会丢失，如果在数据库层面，批量操作在大事务，会导致锁的竞争，并且也会导致主备的不一致。如果是一些不重要的消息如对日志进行备份，就可以使用批量操作之类的提高消费性能，因为一些日志消息丢失也是可以接受的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTg5NTEw_size_1_color_FFFFFF_t_70]: /images/20221119/282a801572d54a12aceb40395b547e30.png