产线问题分析与解决系列：5RocketMQ消息积压问题的解决方案：动态队列分配的应用

程序员（小李）：负责业务功能的开发和优化。 

运维（老张）：负责系统的部署和监控。 

业务方（李总）：负责业务需求和上线进度，强调快速扩张。

场景：会议室中，小李、老张和李总正在讨论美国业务作业高峰期RocketMQ消息堆积告警问题。

李总（焦急地）：小李，老张，这次美国业务作业高峰期的消息堆积问题很严重啊！DBU基础监控频繁告警，消息积压超过30多万条，持续时间长达十几分钟。我们的业务扩张速度这么快，这个问题必须马上解决！

小李（认真地）：李总，我已经深入分析了这个问题。第一次排查时，我们发现消息生产者和消费者处理能力相差太大。在业务高峰期，生产者每秒发送大量消息，但消费者只有2个实例，处理不过来，导致消息积压。我们临时增加了一台服务器，但问题依然存在。

老张（点头）：对，我也监控到了这个问题。第二次排查时，我们登录了RocketMQ控制台，发现topic：canal_sys_scan_history虽然有4个队列，但消息只发给了0号队列，也就是只有1个消费者在消费消息。这解释了为什么增加消费者没有效果。

小李：是的，我们进一步分析了canal的配置文件，发现默认执行分区为0，未开启动态分区，也未设置分区规则。这导致所有消息都集中在0号队列，消费者无法充分利用其他队列。

老张（赞许地）：这个思路不错！我们已经在测试环境中修改了canal的配置文件，启用了动态队列分配，并重启了canal服务。初步效果显示，消息积压问题得到了明显缓解。

李总（稍微放松）：听起来不错，但我们的业务扩张速度很快，数据量会越来越大。这些优化能支撑多久？

小李：李总，这些优化是经过充分测试的，能够支撑当前的数据量。未来如果数据量继续增长，我们可以考虑进一步优化，比如增加消费者实例或者调整RocketMQ的集群配置。

老张：对，我这边也会加强监控，确保系统的稳定性。如果有性能瓶颈，我会第一时间通知小李。

李总（满意地）：好，那你们抓紧时间上线这些优化。我们的业务不能停，消息堆积问题必须尽快解决。

小李：明白，李总。我会尽快完成代码优化和测试，确保不影响业务。

老张：我这边也会配合小李，做好部署和监控工作。

李总（站起身）：好，那就辛苦你们了！希望这次优化能彻底解决问题，让我们的业务继续快速扩张！

产生背景

在美国业务作业高峰期【灾难】xxx基础监控灾难告警，频繁出现消息积压告警，持续时间长达十几分钟.

图片

在监控平台也能查看到队列us_xxx_scan_history消息堆积超过30多万.

图片

分析过程

第一次排查

1. 查看监控平台服务器资源正常、数据库资源正常。分析派送服务6个实例生产消息，msg服务2个实例（4C*8G）消费消息，消息生产者和消费者处理能力相差太大，在业务高峰期时消息处理不过来导致积压。
2. 结论

临时给dbu-mod-msg加了一台服务器,继续观察发现消息处理能力并没有得到提升,消息积压问题依然存在。

第二次排查

1. 拉运维一起登录生产环境RocketMq控制台查看topic队列的配置信息，发现只有1个队列在接收和消费消息。

图片

2. canal监听扫描记录表（sys_scan_history）binlog日志变化，将增、删、改的sql发送消息给rocketmq 。

topic：canal_sys_scan_history，该topic有4个队列,但是控制台显示消息只发给0了号队列,也就是只有1个消费者在消费消息,所以加消费者没有用。

分析canal配置文件找到关键配置信息，发现默认执行分区为：0,未开启动态分区，未设置分区规则。

图片

canal开启消息动态分区。

图片

消费端的配置

图片

没什么特别的配置。

解决之后的监控：

图片

经验总结 

1. 需要了解服务器部署的生产者和消息者资源配置和配比情况。
2. 需要了解整个消息的集成链路。
3. 需要了解业务的实现原理,排除内部干扰。
4. 需要了解开源中间件的底层实现原理和深入了解每个参数配置规则。

相关技术分享

RocketMQ 出现消息积压问题，通常是由生产者、消费者、RocketMQ 自身及外部环境等多方面因素造成的，以下是详细分析：

生产者方面

生产速率过高：生产者发送消息的速度远远超过消费者处理消息的速度。例如，在电商大促期间，短时间内会产生大量的订单消息，生产者高频地将这些消息发送到 RocketMQ，若消费者处理能力不足，就会造成消息积压。

批量发送异常：当生产者采用批量发送消息时，如果批量消息的大小设置不合理，或者网络状况不佳，可能导致批量发送失败。生产者可能会进行重试，这会使消息重复发送，增加了消息的数量，进而引发积压。

生产者配置不合理：如消息发送的线程数设置过多，可能导致 RocketMQ 服务端的负载过高，影响消息的正常处理，使得消息在服务端堆积。

消费者方面

消费能力不足：消费者的处理能力有限，无法及时处理接收到的消息。这可能是因为消费者实例数量过少，或者消费者服务器的硬件资源（如 CPU、内存、磁盘 I/O）不足。例如，一个消费者实例每秒只能处理 100 条消息，但生产者每秒发送 500 条消息，就会造成消息积压。

消费逻辑复杂：消费者的业务逻辑过于复杂，导致处理每条消息的时间过长。例如，在处理消息时需要进行大量的数据库查询、复杂的计算或者与其他系统进行多次交互，都会降低消费的速度。

消费异常处理不当：当消费者在处理消息时出现异常，如果没有正确处理，可能会导致消息不断重试，甚至进入死循环，从而造成消息积压。例如，消费者在处理消息时遇到数据库连接异常，但没有进行合理的重试策略和异常处理，就会使消息一直处于待处理状态。

消费并行度不够：消费者的并行消费能力不足，无法充分利用系统资源。例如，消费者的线程池配置不合理，线程数量过少，不能同时处理多个消息，导致消费速度变慢。

RocketMQ 服务端方面

Broker 性能瓶颈：Broker 是 RocketMQ 的核心组件，负责消息的存储和转发。如果 Broker 的硬件资源不足，如磁盘 I/O 性能低下、内存不足等，会影响消息的读写速度，导致消息积压。此外，Broker 的配置参数不合理，如刷盘策略、缓存大小等，也会影响其性能。

磁盘空间不足：如果 Broker 所在的磁盘空间不足，会导致消息无法正常写入磁盘，从而造成消息积压。当磁盘空间达到一定阈值时，Broker 可能会限制消息的写入，进一步加剧积压问题。

网络问题：RocketMQ 服务端与生产者、消费者之间的网络连接不稳定，会导致消息传输延迟或失败。例如，网络带宽不足、网络丢包等问题，都会影响消息的正常收发，造成消息积压。

集群负载不均衡：在 RocketMQ 集群环境中，如果各个 Broker 节点的负载不均衡，部分 Broker 节点的压力过大，而其他节点的资源利用率较低，就会导致消息在高负载的 Broker 节点上积压。

外部环境方面

依赖服务故障：消费者在处理消息时可能依赖其他外部服务，如数据库、缓存、第三方 API 等。如果这些依赖服务出现故障，会影响消费者的处理速度，导致消息积压。例如，数据库服务出现故障，消费者无法正常将处理结果写入数据库，就会使消息处理受阻。

系统资源竞争：如果 RocketMQ 所在的服务器上还运行着其他高资源消耗的应用程序，会与 RocketMQ 竞争系统资源，如 CPU、内存、磁盘 I/O 等。这会影响 RocketMQ 的性能，导致消息积压。