一、消息积压之困

在使用 RabbitMQ 消息中间件时，消息积压是一个常见且可能影响系统性能和稳定性的问题。消息积压是指生产者发送消息的速度超过消费者处理消息的速度，导致队列中未处理的消息不断增多。

这种情况在高并发、高流量的业务场景中尤为常见。例如，在电商大促活动期间，订单量急剧增加，生产者（可能是订单生成系统）以极高的速度发送消息，而消费者（可能是订单处理系统）的处理能力可能无法及时跟上，从而导致消息在队列中堆积。

消息积压可能会带来以下问题：

1. 系统性能下降：大量未处理的消息占用内存和磁盘空间，可能导致 RabbitMQ 服务器性能下降，影响消息的发送和接收速度。

2. 影响业务处理：如果消息积压严重，可能会导致业务处理延迟，影响用户体验。例如，订单处理延迟可能会导致用户等待时间过长，甚至可能导致用户流失。

3. 增加系统复杂性：为了解决消息积压问题，可能需要采取一些复杂的措施，如增加消费者数量、调整队列配置等，这会增加系统的复杂性和维护成本。

那么，为什么会出现消息积压的情况呢？主要有以下几个原因：

1. 消费者处理速度慢：当消费者处理消息的速度跟不上生产者发送消息的速度时，消息会在队列中堆积。这可能是由于消费者的业务逻辑复杂、耗时长，或者消费者代码性能低，导致处理单个消息的时间过长。

2. 队列配置不合理：队列的长度限制设置不合理，或者队列的持久化设置不当，都可能导致消息积压。例如，如果队列的长度限制设置得太小，当生产者发送消息的速度较快时，队列很快就会被填满，从而导致消息积压。

3. 网络问题：网络延迟或中断可能导致消费者无法及时从 RabbitMQ 服务器获取消息，从而造成消息积压。

4. 服务器配置偏低：在高流量情况下，RabbitMQ 服务器的配置可能不足以处理快速增长的消息量。例如，内存不足、CPU 性能低等都可能导致服务器处理消息的速度变慢，从而引起消息积压。

5. 消费者故障：消费者进程崩溃或处理逻辑错误，导致无法继续消费消息。

6. 程序逻辑设计问题：生产者持续生产消息，而消费者由于逻辑设计不当，无法有效消费消息。例如，生产者没有考虑消费者的处理能力，盲目发送消息，或者消费者在处理消息时出现死循环等问题。

二、积压原因剖析

1. 消费者宕机，消息无法及时被消费。

当消费者宕机时，消息队列中的消息无法被及时消费，从而出现积压。这种情况可能是由于消费者进程崩溃或处理逻辑错误导致无法继续消费消息。例如，在系统运行过程中，消费者所在的服务器出现硬件故障或软件错误，导致消费者进程意外终止，此时队列中的消息就无法被处理，逐渐堆积起来。

2. 消费者逻辑处理耗时，消费能力不足。

消费者在处理消息时，如果本身逻辑处理数据耗时较长，就会导致消费能力不足，引起队列消息积压。比如，消费者在处理消息时需要进行复杂的数据库查询、网络请求或大量的计算操作，这些操作可能会耗费较长的时间，使得消费者处理单个消息的速度变慢，无法跟上生产者发送消息的速度。

3. 生产方消息产生过多，如 “双 11 大促活动”。

在一些特定的场景下，如 “双 11 大促活动”，消息生产方单位时间内产生的消息过多，而消费者的处理能力有限，导致消费者处理不过来，从而出现消息积压。在这种高流量的情况下，生产者（如订单生成系统）以极高的速度发送消息，而消费者（如订单处理系统）可能无法及时处理如此大量的消息，使得队列中的未处理消息不断增多。

三、应对之策

1. 增加消费者数量，提高消息处理速度。可启动更多工作进程或线程，上线更多消费者，或增加服务器数量。

通过增加消费者的数量来提高消息的处理速度是一种常见的应对消息积压的方法。例如，可以在消费者端配置多个并发消费者，在 Spring AMQP 中可以设置 SimpleMessageListenerContainer 的 concurrentConsumers 属性。也可以像在某些代码示例中那样，启动更多的工作进程或线程来创建消费者，如在ConsumerWorker类中，通过在main方法中启动多个消费者线程来处理积压的消息。同时，可以根据监控系统的告警动态地调整消费者的数量，或者在系统设计时就允许消费者自动扩展，例如通过 RabbitMQ 的自动扩展消费者数量的配置，在SimpleMessageListenerContainer类中，可以设置启动新的消费者最小时间间隔、停止空闲消费者最小时间间隔等参数，根据消费者的连续成功消费和连续空闲情况来自动调整消费者数量。

2. 优化消费者逻辑，减少不必要计算和数据库查询优化等，提高单个消费者处理效率。

对消费者的业务逻辑进行分析和优化可以提高单个消费者的处理效率。这可能涉及算法优化、数据库访问优化或者缓存机制的使用等。例如，减少不必要的计算操作，避免复杂的数据库查询，合理使用缓存来提高数据的访问速度。同时，要确保消费者有足够的 CPU、内存和网络资源来处理消息，避免由于资源限制导致消费速度慢。合理处理消息消费过程中的异常，确保不会因为单个消息的处理问题导致整个消费进程崩溃。

3. 调整预取计数，设置适当的 basic.qos 值限制每个消费者同时处理的消息数量。

重置监听器中的预取计数是指在使用 RabbitMQ 和 Spring AMQP 时，可以通过修改监听器的预取计数来控制消费者从队列中预先获取的消息数量。通过调整预取计数，可以灵活地控制消费者的并发处理能力和消息处理的效率。在 RabbitMQ 中，预取计数的默认值为 1，即每次只预取一条消息。可以增大预取计数来提高消费者的并发处理能力，加快消息处理的速度，但过大的预取计数可能会导致消费者一次性获取过多的消息，造成资源浪费和消息处理的不均衡。在 Spring AMQP 中，可以通过设置 prefetchCount 属性来修改预取计数。

4. 正确使用消息确认机制，消费者成功处理消息后及时发送确认信号给 RabbitMQ。

RabbitMQ 的消息确认机制分为生产者确认机制和消费者确认机制。消费者接收到消息后，需要向 RabbitMQ 服务器发送确认消息，以告诉服务器已经成功地接收并处理了该消息。如果消费者没有发送确认消息，RabbitMQ 服务器会认为该消息没有被正确地处理，从而会将该消息重新发送给其他消费者进行处理。在编码时可以用两个选项用来控制消息投递的可靠性：消息从 producer 到 RabbitMQ broker cluster 成功，则会返回一个 confirmCallback；消息从 exchange 到 queue 投递失败，则会返回一个 returnCallback。可以利用这两个 callback 接口来控制消息的一致性和处理一部分的异常情况。

5. 设置消息过期时间，对不需要长期保留的消息自动丢弃。

对于那些不需要长期保留的消息，可以为它们设置一个过期时间（Time-To-Live, TTL），超过这个时间后消息将自动被丢弃。这样可以避免消息在队列中长时间堆积，占用内存和磁盘空间。

6. 使用死信交换器，将未能处理的消息转移到死信队列后续处理。

当消息在队列中达到最大重试次数或过期时，它会被转发到死信交换器。在 RabbitMQ 中配置备份交换机的步骤包括定义备份交换机和队列，例如可以通过设置队列的参数，将未能处理的消息转移到死信队列中，以便后续分析或重试处理。在消费者中可以处理从备份交换机接收到的消息，如记录日志或重试。

7. 监控与报警，持续跟踪队列长度和消息速率等关键指标，及时预警。

设置监控系统来持续跟踪队列长度、消息速率等关键指标，并在达到预警阈值时发出警报，以便及早采取行动。可以使用 RabbitMQ 管理插件或其他监控工具来持续监控系统的性能指标，如消息速率、队列长度、CPU 和内存使用情况等。定期审查日志文件，寻找可能的性能瓶颈或错误。

8. 水平扩展 RabbitMQ 集群，增加节点提升整体吞吐量。

在高并发场景下，考虑横向扩展 RabbitMQ 集群，通过增加节点来提升整体吞吐量。将多个 RabbitMQ 节点组成集群，实现负载均衡和故障转移。确保集群中的节点分布在不同的物理主机上，以防止单一故障点。

9. 根据业务需求为消息设定优先级，确保重要消息优先处理。

根据业务需求为不同消息设定优先级，确保重要的消息能够优先得到处理。例如，可以对重要消息设置较高的优先级，使其能够被更快地消费。

10. 实施流量控制，限制生产者发送速率或暂停部分非紧急任务消息产生。

实施流量控制措施，比如限制生产者的发送速率，或者在检测到堆积严重时暂停部分非紧急任务的消息产生。可以对生产者的发送速度进行限流，避免消息过快地进入队列。也可以根据实际情况，暂停部分非紧急任务的消息产生，以减轻消费者的压力。

四、特殊情况处理

1. 积压严重时的快速处理

当消息积压严重时，我们可以采取以下步骤进行快速处理：

• 修改消费者问题，确保恢复消费速度后停掉现有消费者：首先，排查并修复消费者端的问题，使其能够以正常的速度消费消息。一旦消费者恢复了消费速度，将现有的消费者都停掉，为后续的临时扩容做准备。

• 新建 topic 和更多队列，编写临时分发数据程序：临时建立一个新的 topic，其 partition 数量可以是原来的 10 倍。同时，建立好原先 10 倍或者 20 倍的 queue 数量。然后，编写一个临时的分发数据的 consumer 程序，这个程序部署上去后，负责消费积压的数据。消费之后不做耗时的处理，直接均匀轮询写入临时建立好的 10 倍数量的 queue。

• 征用更多机器部署消费者，快速消费积压数据后恢复原架构：接着，临时征用 10 倍的机器来部署 consumer，每一批 consumer 消费一个临时 queue 的数据。这样做相当于临时将 queue 资源和 consumer 资源扩大 10 倍，以正常速度的 10 倍来消费消息。等快速消费完积压数据之后，恢复原来的部署架构，重新用原先的 consumer 机器来消费消息。

2. 消息过期失效处理

如果消息因过期被清理，我们可以采取以下措施进行处理：

• 写临时程序查出丢失数据并灌入 MQ 补回：在这种情况下，实际上没有什么消息积压，而是丢了大量的消息。所以第一种增加 consumer 肯定不适用。这种情况可以采取 “批量重导” 的方案来进行解决。在流量低峰期（比如夜深人静时），写一个程序，手动去查询丢失的那部分数据，然后将消息重新发送到 mq 里面，把丢失的数据重新补回来。

五、其他方法探索

1. 增加处理能力，如优化系统架构、增加服务器资源和采用负载均衡。

通过优化系统架构、增加服务器资源以及采用负载均衡等手段，可以显著提高系统的处理能力和并发处理能力。例如，可以对系统进行性能调优，去除不必要的中间环节，使消息的流转更加高效。增加服务器资源可以包括增加内存、CPU 核心数等，确保系统能够及时处理所有的消息。采用负载均衡可以将消息均匀地分配到多个服务器上，避免单个服务器负载过高。

2. 异步处理消息，接收消息后立即返回响应再后续处理。

将消息的处理过程设计为异步执行是一种有效的方式。当接收到消息后，立即返回响应，然后将消息放入队列中进行后续处理。这样可以避免同步请求的阻塞，提高系统的吞吐量和响应速度。例如，在 Web 应用中，用户提交一个请求后，系统可以立即返回一个确认信息，然后在后台异步处理该请求，避免用户长时间等待。

3. 消息分片，将大或复杂消息拆分为小部分处理。

如果消息体较大或者复杂，可以考虑将消息分片处理。将消息拆分为多个小的部分进行处理，减少单个消息的处理时间，从而提高整体处理能力。例如，对于一个大型文件的传输，可以将文件分割成多个小块，分别发送到消息队列中，由消费者逐个处理这些小块，最后再合并成完整的文件。

4. 集群扩展，添加更多节点实现分布式部署和负载均衡。

根据实际情况，可以考虑通过添加更多的节点来扩展消息处理的集群规模，实现分布式部署和负载均衡，以应对大量消息的处理需求。例如，可以将多个 RabbitMQ 节点组成集群，每个节点负责处理一部分消息，从而提高整体的处理能力。同时，通过负载均衡算法，可以将消息均匀地分配到各个节点上，避免单个节点负载过高。

5. 优化数据库操作，减少数据库压力。

如果消息的处理涉及到数据库操作，可以考虑对数据库查询和写入进行性能优化，如建立索引、合理使用缓存等，以减少数据库的压力。例如，对于频繁查询的字段，可以建立索引，提高查询速度。合理使用缓存可以避免重复查询数据库，提高数据的访问速度。

6. 增加消费者数量和队列数量。

增加消费者数量可以提高消息处理速度，从而减少消息积压。可以根据消息的类型和优先级分配消费者，使消息得到及时处理。同时，增加队列数量也可以缓解消息积压。根据消息的类型和优先级，可以将不同类型的消息存储在不同的队列中，更好地管理消息流量。

7. 设置消息过期时间和使用限流机制。

可以设置消息的过期时间，当消息在队列中等待时间超过指定时间时，会被自动删除，直到消息被正确处理或超过最大重试次数为止。同时，使用限流机制可以控制消费者的消费速度，避免消息过多导致消费者无法及时处理。可以使用 Qos 机制，设置每个消费者同时处理消息的最大数量，从而保证系统的性能和稳定性。

Tags：rabbitmq配置