消息队列的积压

消息队列的积压问题个人认为分为两种
一种是消费者跟不上生产者的速度,生产者只是投递消息,而消费者是有业务的,会操作数据库等操作.当然生产者生产消息也是根据业务来讲,不可能死循环来投递消息.
那如何保证生产者和消费者之间的一个平衡.
(1) 消费者:可以增加消费者的数量,可以快速的消费消息,最简单的方法就是增加服务器
(2) 生产者:放缓生产者投递消息的频率

1. 线程睡眠的方式,这种方式不推荐,如果线程睡眠,肯定会创建大量的线程,增加生产者的压力
2. 定义一个全局变量记录上一次的生产时间,和当先时间比较,大于消费者的消费时间才去投递消息,这样涉及到线程安全问题,简单的使用volatile修饰上一次的生产时间是不行的,因为比较时间,并赋值全局变量生产时间不是原子操作,volatile不支持原子性,只能使用锁,如果是分布式项目要使用分布式锁,降低了生产者效率.
3. 消息的过期时间,当消息过期之后放到死信队列中,rabbitmq和rocketmq都可以设置过期时间,并达到过期时间之后放到死信队列中,设置一个专门消息者来消费死信队列的消费者,可以记录日志人为的干预
   另一种是突发情况,消费者操作数据库宕机了,或者就是消费者本身宕机了.
   这种情况的如何解决呢
   Rabbitmq的解决方式
   (1) 解决消费者不能消费消息的原因,比如恢复数据库等操作.
   (2) 可以添加多个消费者,去消费这个队列,因为rabbitmq的队列可以被多个消费者消费,不用担心队列的速度,因为队列效率要高于消费者,所以可以不增加队列.
   Rocketmq的解决方式,如果大量的消息积压,是不能通过增加消费者来解决的,因为如果一个topic创建了4个队列(默认),那一个消费者组超过4个消费者,多出去的消费者是闲置状态,除非正在消费的消费者宕机了20s后,多出去的消费者会开始消费消息.
   那针对rocketmq如何解决积压消息
   (1) 解决消费者不能消费消息的原因,比如恢复数据库等操作.
   (2) 新添加消费者组,订阅已经积压的topic ,消费者组拉取的消息,作为生产者重新投递到一个新的topic,这个topic就是步骤1重新订阅的topic
   (3) 新的topic可以多创建队列.
   (4) 增加原有消费者的个数,订阅新的topic,这样就可以多个消费者同时消费消息.
   那会不会有疑问,为什么不在原有的topic,扩容队列并创建对应的消费者呢,因为扩容队列不会将原有队列的消息分摊给新的消息队列,扩容是生产新的消息会分发到扩容队列中.