消息堆积

[TOC]

一、消息堆积主要原因

消息堆积主要原因：

1、生产者的生产消息速度>>消费者的处理消息速度，速度不匹配从引起的堆积；（消费者活着但是处理慢）

2、消费者实例IO阻塞严重或者挂机；（消费者宕机等）

3、消费者故障期间消息的堆积。（堆积累加）

单从增加消费者数是远远不够。之所以要处理消息堆积，是为了防止消息堆积所引起MQ的异常，所以在所有MQ的业务场景，消息如果是重要的，不容丢弃时，需要有备选方案，可以采用数据转移，增加中间缓冲技术。

二、不同消息中间件面对消息堆积的能力理解

2.1 RocketMQ消息堆积

RocketMQ在介绍消息堆积能力时，介绍如下：

除了异步解耦功能，消息队列 RocketMQ 版还有挡住前端数据洪峰的重要功能，以此保证后端系统的稳定性。这要求消息队列 RocketMQ 版具有一定的消息堆积能力。消息队列 RocketMQ 版能支持 10 亿级别的消息堆积，不会因为消息堆积导致性能明显下降。

---阿里云官网 如何处理消息堆积？

2.2 RabbitMQ消息堆积

而RabbitMQ在介绍优缺点时，消息堆积作为缺点之一：

RabbitMQ 对消息堆积的支持并不好，在它的设计理念里面，消息队列是一个管道，大量的消息积压是一种不正常的情况，应当尽量去避免。当大量消息积压的时候，会导致 RabbitMQ 的性能急剧下降

延伸一：为什么说RabbitMQ，对消息堆积的支持并不好？

image.png

2.3 从存储模型来理解

关于消息队列对于消息堆积的堆积能力，还需要从消息队列的存储模型来分析：

• 1、 RabbitMQ：内存、磁盘都保存，消息保存到内存中，通过镜像队列保证HA，通过磁盘存储保证持久化。但由于内存队列中需要保存所有完整的消息本地，因此当消息堆积太多时，会使得内存空间不可用，严重可能内存溢出，服务宕机。

​ -----即 内存、磁盘，支持少量堆积

• 2、 RocketMQ：消息持久化保存到磁盘中，且消费队列本身不保存消息本地，保存消息磁盘索引，通过FileChannel的MMAP机制实现内存映射，处理消息时能达到基本和内存相同的效率。设置同步复制和同步刷盘即可保存消息不丢失。

​ -----即 磁盘+内存映射技术，支持大量堆积。【磁盘空间还是足够富裕的】

• 3、 Kafka：同RocketMQ。

附：RocketMQ存储模型图如下：

image.png

三、如何处理消息堆积

如何处理消息堆积呢？可以从两个当面考虑：

• 如何通过优化代码来避免消息堆积
• 消息已经堆积了，线上如何快速处理

3.1 如何预防消息堆积

在消息的收发两端，我们的业务代码怎么和消息队列配合，达到一个最佳的性能。

1、发送端性能优化

从消息堆积若干原因来看，消息堆积的原因主要在消费端处理上，本身生产者端应该遵循的原则应该是尽可能快的将消息发送到Broker中去，因此发送端除了业务处理时批量发送暂无好的手段优化，而且并不是所有的业务处理都支持批量发送和批量接收处理。

发送端业务代码的处理性能，实际上和消息队列的关系不大，因为一般发送端都是先执行自己的业务逻辑，最后再发送消息。如果说，你的代码发送消息的性能上不去，你需要优先检查一下，是不是发消息之前的业务逻辑耗时太多导致的。

• 批量发送是发送端预防消息堆积的方式之一。

2、消费端性能优化

在设计系统的时候，一定要保证消费端的消费性能要高于生产端的发送性能，这样的系统才能健康的持续运行。

• 方式1 增加单个消费者处理能力

  增加单个消费者的处理能力这块没有绝对的办法，只能尽可能的优化消息处理业务逻辑的能力，减少不必要的非业务相关处理时间消耗；如果消息处理业务已经优化到无法再优化了，那只能通过方式2水平扩展消费者个数来优化。

• 注意：部分同学采用在业务处理OnMessage时，先将消息保存到内存队列中，再开启线程池并发处理内存队列缓存消息这种方式（即通过内存队列增加一个异步环节）-----这种方式存在丢消息的风险，如果消费节点宕机，内存队列中的消息直接丢失。慎用这种方式。

• 方式2 水平扩容消费者个数

消费端的性能优化除了优化消费业务逻辑以外，也可以通过水平扩容，增加消费端的并发数来提升总体的消费性能。

注意：水平扩容是应保证 扩容后消费者个数<=分区或者队列个数

反过来，即如果扩容后消费者个数超过分区或者队列个数后，再扩容已经没有意义。---因为单个消费队列同一时间内只能被一个消费者消费，再多的消费者也没有用。

此时，需要在Broker中同步增加分区或者队列个数，扩容消费者才有意义。

补充：Kafka中叫分区Partition，RocketMQ和RabbitMQ中叫队列Queue

3.2 消息已经堆积，如何快速处理

如果消息已经堆积了，线上如何快速处理。对于系统发生消息积压的情况，需要先解决积压，再分析原因，毕竟保证系统的可用性是首先要解决的问题。

快速解决积压的方法就是通过水平扩容增加 Consumer 的实例数量，以及其他方式如下。

• 1、消费端扩容；--通用方式
• 2、服务降级；--快速失败，不一定适用所有业务场景
• 3、异常监控。--属于运维层面措施

同步文章见同步博客地址

附 参考文章

1、阿里云官网 如何处理消息堆积？