深入浅出 RocketMQ 消息队列 笔记（13）

如何处理消息堆积问题 消息堆积的本质：消费速度<生产速度 常见消息堆积原因 1.瞬时流量 2.消费性能不足 3.机器不够 4.Bug 5.其他功能影响 瞬时流量 比如平常一分钟一条消息，瞬时流量来了，可能一秒钟就产生了上万条消息。 这里要对消息的瞬时性做分析，如果只是瞬时的，持续几分钟就好了，那就不需要做任何改动，毕竟消息队列就是负责削峰填谷的。如果持续时间长且消息量级过大，那堆积造成的影响还是需要重视。 这里举个例子，是yes哥之前遇到的一个情况，有七千万的历史数据需要清洗（消费），其中生产者花了12天时间将数据发给消息队列，但消费者一天只能消费200w条消息，所以乐观估计，消费者消费完所有历史数据需要至少35天。 这里会带来不少问题，如果消息队列用的是第三方提供的（比如阿里云），消息存储有个默认时间3天，如果消息存储3天还没被消费，那这个消息就会被删掉了（删掉的目的是为了控制存储成本），堆积未消费的消息丢失后，还需要生产者重新发送一遍，会很麻烦 还有一个问题，如果历史数据跟正常业务数据的处理流程是相同的，也就是他们的topic相同，那在broker产生堆积后，光顾着处理历史数据消息了，正常业务数据的处理也会因此而堵住了。 这里yes哥的解决方式是降低生产消息的频率，不持续发消息，让消息发送的频率略低于消费速率，给正常业务数据有消费的空间 消费性能问题 1.可以把循环里的查询拎出来放在外面批量查询，然后Map映射 2.单条插入/更新改成批量插入/更新 机器不够 消费性能上能优化还是优化了，但消息还是堆积，那就得水平扩展机器了，不过要注意，消费者与队列数要一直保持消费<队列的数量，否则会造成消费者空忙，不能很好的利用重平衡机制。 Bug 也有可能是消费逻辑有bug，导致消息堆积。消费失败需要重试，默认重试16次才会进入死信队列，相当于一条消息用到的流量被放大了16倍，资源都用来重试了，重试后还是失败。再如果消息是顺序消息，那后面的消息都处理不下去了。这种情况只能对消息队列做监控，发现报错后紧急发布版本，修复bug。进入死信队列的消息可以手动重新消费。 其他功能影响 其他业务功能抢夺了消息队列在使用的资源，比如消息消费的逻辑是扣减库存，其他业务功能也涉及到扣减库存，那么就有可能会争夺同一个商品的处理，也就是同一个数据库的行锁，导致消费速率降低。还有可能是其他功能的长事务导致消费等待，等等。如果平常正常，流量也没有变很大，有天突然堆积了，可以在这方面考虑问题原因。