RabbitMQ消息积压

消息积压

线上有时因为发送方发送消息速度过快，或者消费方处理消息过慢，可能会导致broker积压大量未消费消息。消息数据格式变动或消费者程序有bug，导致消费者一直消费不成功，也可能导致broker积压大量未消费消息。解决方案可以修改消费端程序，让其将收到的消息快速转发到其他主题，可以设置很多分区，然后再启动多个消费者同时消费新主题的不同分区。可以将这些消费不成功的消息转发到其它队列里去，类似死信队列，后面再慢慢分析死信队列里的消息处理问题。另外在RocketMQ官网中，还分析了一个特殊情况，如果RocketMQ原本是采用的普通方式搭建主从架构，而现在想要中途改为使用Dledger高可用集群，这时候如果不想历史消息丢失，就需要先将消息进行对齐，也就是要消费者把所有的消息都消费完，再来切换主从架构。因为Dledger集群会接管RocketMQ原有的CommitLog日志，所以切换主从架构时，如果有消息没有消费完，这些消息是存在旧的CommitLog中的，就无法再进行消费了。这个场景下也是需要尽快的处理掉积压的消息。

延迟队列

消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费。例如10分钟，内完成订单支付，支付完成后才会通知下游服务进行进一步的营销补偿。往MQ发一个延迟1分钟的消息，消费到这个消息后去检查订单的支付状态，如果订单已经支付，就往下游发送下单的通知。而如果没有支付，就再发一个延迟1分钟的消息。最终在第10个消息时把订单回收，就不用对全部的订单表进行扫描，而只需要每次处理一个单独的订单消息。这个就是延迟对列的应用场景。rabbittmq,rocketmq都可以通过设置ttl来设置延迟时间，kafka则是可以在发送延时消息的时候，先把消息按照不同的延迟时间段发送到指定的队列中，比如topic_1s，topic_5s，topic_10s，topic_2h，然后通过定时器进行轮训消费这些topic，查看消息是否到期，如果到期就把这个消息发送到具体业务处理的topic中，队列中消息越靠前的到期时间越早，具体来说就是定时器在一次消费过程中，对消息的发送时间做判断，看下是否延迟到对应时间了，如果到了就转发，如果还没到这一次定时任务就可以提前结束了。

mq设置过期时间，就会有消息失效的情况，如果消息在队列里积压超过指定的过期时间，就会被mq给清理掉，这个时候数据就没了。解决方案也有手动写程序，将丢失的那批数据，一点点地查出来，然后重新插入到 mq 里面去。