RabbitMQ、kafaka、rocketmq等消息队列MQ消息堆积如何解决

本文主要是介绍RabbitMQ、kafaka、rocketmq等消息队列MQ消息堆积如何解决，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

概述

1.产生背景： 生产者投递消息的速率与我们消费者消费的速率完全不匹配。
2.生产者投递消息的速率>消费者消费的速率
导致我们消息会堆积在我们 mq 服务器端中，没有及时的被消费者消费 所以就会产生消息堆积的问题
3.注意的是：rabbitmq 消费者我们的消息消费如果成功的话 消息会被立即删除（自动ack）
kafka 或者 rocketmq 消息消费如果成功的话，消息是不会立即被删除。
4.解决办法：
A.提高消费者消费的速率；（对我们的消费者实现集群）
B.消费者应该批量形式获取消息 减少网络传输的次数；

解决方案

消息堆积是消息队列系统中常见的问题，可能会导致系统性能下降、延迟增加甚至消息丢失。下面是一些解决 RabbitMQ、Kafka、RocketMQ 等消息堆积问题的方法：

1. 监控和预警：
   设置监控指标，定期监控消息队列中消息积压情况，如消息堆积量、消费者处理速度等。当消息堆积超过阈值时，发送预警通知，及时发现问题并采取措施。
2. 扩展消费者：
   增加消费者数量，提升消息处理速度，以缓解消息堆积问题。可以动态地增加消费者实例来分担消息处理压力，确保消息能够及时被处理。
3. 优化消费者端处理逻辑：
   检查消费者端的处理逻辑是否高效合理，避免因为消费者处理速度慢导致消息堆积。优化消费者代码，减少不必要的处理时间，提高消息处理效率。
4. 调整消息消费方式：
   根据业务需求和场景，调整消息消费方式，如批量消费、并发消费等。合理利用消息队列提供的特性，提高消息消费效率，减少消息堆积发生的可能性。
5. 调整消息队列参数：
   根据实际情况，调整消息队列的参数设置，如队列长度、超时时间等。合理设置参数可以更好地适应业务需求，防止消息堆积问题的发生。
6. 消息重试机制：
   实现消息重试机制，当消息处理失败时，自动进行重试，确保消息最终被正确处理。避免因为消息处理失败而导致消息堆积问题。
7. 定时清理历史数据：
   定期清理历史数据，删除过期和无用的消息，释放存储空间，避免消息堆积。保持消息队列的数据清洁，有助于提高系统的性能和稳定性。

消息堆积如何处理

主要是消息的消费速度跟不上生产速度，从而导致消息堆积。解决思路：
1.可能是刚上线的业务，或者大促活动，流量评估不到位，这时需要增加消费组的机器数量，提升整体消费能力
2.也可能是消费端的问题，正常情况，一条消息处理需要10ms，但是优化不到位或者线上bug，现在要500ms，那么消费端的整体处理速度会下降50倍。这时，我们就要针对性的排查业务代码。例如数据库的一条sql没有命中索引，导致单条消息处理耗时拉长，进而导致消息堆积

如果 bug 导致几百万消息持续积压几小时。有如何处理呢？ 需要解决bug，临时紧急扩容，大概思路如下：
1.先修复 consumer 消费者慢问题，以确保其恢复消费速度，然后将现有consumer 都停掉。
2.新建一个 topic，partition 原来 10 倍，临时立好原先 10 倍 queue数量。
3.然后写一个临时分发数据consumer 程序，这个程序部署上去消费积压数据，消费之后不做耗时处理，直接均匀轮询写入临时立好10 倍数量queue。
4.接着临时征用 10 倍机器来部署 consumer，每一批 consumer 消费一个临时queue 数据。这种做法相当于临时将 queue 资源和 consumer 资源扩大10 倍，以正常 10 倍速度来消费数据。
5.等快速消费完积压数据之后，得恢复原先部署架构，重新用原先consumer 机器来消费消息。

如何解决消息队列的延时以及过期失效问题？消息队列满了以后该怎么处理？有几百万消息持续积压几小时，怎么办？

消息积压处理办法：临时紧急扩容先修复 consumer 的问题，确保其恢复消费速度，然后将现有 cnosumer 都停掉。新建一个 topic，partition 是原来的 10 倍，临时建立好原先 10 倍的 queue 数量。然后写一个临时的分发数据的 consumer 程序，这个程序部署上去消费积压的数据，消费之后不做耗时的处理，直接均匀轮询写入临时建立好的 10 倍数量的 queue。接着临时征用 10 倍的机器来部署 consumer，每一批 consumer 消费一个临时 queue 的数据。这种做法相当于是临时将 queue 资源和 consumer 资源扩大 10 倍，以正常的 10 倍速度来消费数据。等快速消费完积压数据之后，得恢复原先部署的架构，重新用原先的 consumer 机器来消费消息。 MQ中消息失效：假设你用的是 RabbitMQ，RabbtiMQ 是可以设置过期时间的，也就是 TTL。如果消息在 queue 中积压超过一定的时间就会被 RabbitMQ 给清理掉，这个数据就没了。那这就是第二个坑了。 这就不是说数据会大量积压在 mq 里，而是大量的数据会直接搞丢。我们可以采取一个方案，就是批量重导，这个我们之前线上也有类似的场景干过。就是大量积压的时候，我们当时就直接丢弃数据了，然后等过了高峰期以后，比如大家一起喝咖啡熬夜到晚上12点以后，用户都睡觉了。这个时候我们就开始写程序，将丢失的那批数据，写个临时程序，一点一点的查出来，然后重新灌入 mq 里面去，把白天丢的数据给他补回来。也只能是这样了。假设 1 万个订单积压 在 mq 里面，没有处理，其中 1000 个订单都丢了，你只能手动写程序把那 1000 个订单给查出来，手动发到 mq 里去再补一次。 mq消息队列块满了：如果消息积压在 mq 里，你很长时间都没有处理掉，此时导致 mq 都快写满了，咋办？这个还有别的办法吗？没有，谁让你第一个方案执行的太慢了，你临时写程序，接入数据来消费，消费一个丢弃一个，都不要了，快速消费掉所有的消息。然后走第二个方案，到了晚上再补数据吧。

这篇关于RabbitMQ、kafaka、rocketmq等消息队列MQ消息堆积如何解决的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---