Rabbitmq消息堆积的问题以及解决方案_springboot rabbitmq 消息堆积

1.消息堆积的原因：

短时间内出现了大量的数据涌入

消费速度慢

2.解决方案

2.1 预防措施：

生产者：

减少发布频率，考虑使用队列最大长度限制，尽可能保证消息都可以被接收。

消费者：

• 增加消费者的处理能力，优化代码，使执行时间更快，使用jdk的队列缓存数据，多线程去处理（要考虑顺序，要不然就发生消费顺序的问题了）。
• rabbitmq消费者为单线程串行消费（org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer类的concurrentConsumers与txSize（对应prefetchCount）都是1），设置并发消费两个关键属性concurrentConsumers和prefetchCount。concurrentConsumers：设置的是对每个listener在初始化的时候设置的并发消费者的个数；prefetchCount：每次从broker里面取的待消费的消息的个数。

综合：

• 生产者端缓存数据，在mq被消费完后再发送到mq，打破发送循环条件。设置合适的qos值（channel.BasicQos()方法：每次从队列拉取的消息数量），当qos值被用光，而新的ack没有被mq接收时，就可以跳出发送循环，去接收新的消息。
• 消费者主动block接收进程，消费者感受到接收消息过快时主动block，利用block和unblock方法调节接收速率，当接收线程被block时，跳出发送循环。

2.2 事故发生

简单修复

修复consumer的问题，让他恢复消费速度，然后等待几个小时消费完毕

复杂修复

先修复consumer的问题，确保其恢复消费速度
2）新建一个topic，partition是原来的10倍，临时建立好原先10倍或者20倍的queue数量
3）然后写一个临时的分发数据的consumer程序，这个程序部署上去消费积压的数据，消费之后不做耗时的处理，直接均匀轮询写入临时建立好的10倍数量的queue
4）接着临时征用10倍的机器来部署consumer，每一批consumer消费一个临时queue的数据
5）这种做法相当于是临时将queue资源和consumer资源扩大10倍，以正常的10倍速度来消费数据
6）等快速消费完积压数据之后，得恢复原先部署架构，重新用原先的consumer机器来消费消息