常见问题总结

MQ常见问题

消息队列核心解决的问题主要是:异步、解耦、消息切峰。异步、解耦、消峰填谷这是消息队列最大的优点,除了这些消息队列还可以会解决一些我们特殊业务场景的问题。但是缺点主要在于系统的可用性、复杂性、一致性问题,引入消息队列后,需要考虑MQ的可用性,万一MQ崩溃了岂不是要爆炸?而且复杂性明显提高了,需要考虑一些消息队列的常见问题和解决方案,还有就是一致性问题,一条消息由多个消费者消费,万一有一个消费者消费失败了,就会导致数据不一致。

RabbitMQ

  • 单机吞吐量:万级别
  • 失效性:微秒级别
  • 可用性:基于主从架构

RabbitMQ现在使用的较为多一些,社区活跃度也很高,功能也很强大,官方还提供了管理的web界面,性能也很好,但是RabbitMQ性能好的主要原因是因为使用erlang语言开发的,erlang语言貌似天生性能好,但对于我们java开发者来说,源码基本看不懂,更别提深入的研究了,不过spring推出了rabbit的支持,貌似还比较好用,比自己去封装实现并且去处理一些问题的要好多了。

RabbitMQ 模式

  • 单机模式

    单机模式通常是用来进行测试和开发的场景,测试一般是否能正确的处理数据,线上环境没人去用单机模式,风险大。

  • 普通集群模式

    普通集群模式就是启动多个 RabbitMQ 实例,在你创建queue只会放在一个 RabbitMQ上,但是每个实例都同步 queue 的元数据,在消费的时候,如果连接到了另外一个实例,那么那个实例会从queue所在实例上拉取数据过来。这种方式确实很麻烦,也不怎么好,没做到所谓的分布式,就是个普通集群。因为这导致你要么消费者每次随机连接一个实例然后拉取数据,要么固定连接那个queue所在实例消费数据,后者有数据拉取的开销,前者导致单实例性能瓶颈。而且如果那个放queue的实例宕机了,会导致接下来其他实例就无法从那个实例拉取,如果你开启了消息持久化,让RabbitMQ落地存储消息的话,消息不一定会丢,得等这个实例恢复了,然后才可以继续从这个queue拉取数据。这方案主要是提高吞吐量的,就是说让集群中多个节点来服务某个queue的读写操作

  • 镜像集群模式

    镜像集群模式是所谓的RabbitMQ的高可用模式,跟普通集群模式不一样的是,你创建的queue,无论元数据还是queue里的消息都会存在于多个实例上,然后每次你写消息到queue的时候,都会自动把消息到多个实例的queue里进行消息同步。

    优点在于你任何一个实例宕机了,没事儿,别的实例都可以用。缺点在于性能开销太大和扩展性很低,同步所有实例,这会导致网络带宽和压力很重,而且扩展性很低,每增加一个实例都会去包含已有的queue的所有数据,并没有办法线性扩展queue。

    开启镜像集群模式可以去RabbitMQ的管理控制台去增加一个策略,指定要求数据同步到所有节点的,也可以要求就同步到指定数量的节点,然后你再次创建queue的时候,应用这个策略,就会自动将数据同步到其他的节点上去了。

如何保证消息的幂等性

消息重复消费的原因主要是在与回馈机制,在某些场景中我们采用回馈机制不同,原因也不同,比如消费者消费完成消息之后回复 ack ,但是刚消费完成还没有来的及提交,系统就重新启动。这时重新启动就会pull 消息的时候没有提高 ack ,消息还是上次的消息。

那么如何怎么来保证消息消费的幂等性呢?实际上我们只要保证多条相同的数据过来的时候只处理一条或者说多条处理和处理一条造成的结果相同即可,但是具体怎么做要根据业务需求来定,例如入库消息,先查一下消息是否已经入库啊或者说搞个唯一约束啊什么的,还有一些是天生保证幂等性就根本不用去管,例如redis就是天然幂等性。

还有一个问题,消费者消费消息的时候在某些场景下要放过消费不了的消息,遇到消费不了的消息通过日志记录一下或者搞个什么措施以后再来处理,但是一定要放过消息,因为在某些场景下例如spring-rabbitmq的默认回馈策略是出现异常就没有提交ack,导致了一直在重发那条消费异常的消息,而且一直还消费不了,这就尴尬了,后果你会懂的。

消息遗漏

1)生产者弄丢了数据
  生产者将数据发送到RabbitMQ的时候,可能数据就在半路给搞丢了,因为网络啥的问题,都有可能。此时可以选择用RabbitMQ提供的事务功能,就是生产者发送数据之前开启RabbitMQ事务(channel.txSelect),然后发送消息,如果消息没有成功被RabbitMQ接收到,那么生产者会收到异常报错,此时就可以回滚事务(channel.txRollback),然后重试发送消息;如果收到了消息,那么可以提交事务(channel.txCommit)。但是问题是,RabbitMQ事务机制一搞,基本上吞吐量会下来,因为太耗性能。

  所以一般来说,如果你要确保说写RabbitMQ的消息别丢,可以开启confirm模式,在生产者那里设置开启confirm模式之后,你每次写的消息都会分配一个唯一的id,然后如果写入了RabbitMQ中,RabbitMQ会给你回传一个ack消息,告诉你说这个消息ok了。如果RabbitMQ没能处理这个消息,会回调你一个nack接口,告诉你这个消息接收失败,你可以重试。而且你可以结合这个机制自己在内存里维护每个消息id的状态,如果超过一定时间还没接收到这个消息的回调,那么你可以重发。

  事务机制和cnofirm机制最大的不同在于,事务机制是同步的,你提交一个事务之后会阻塞在那儿,但是confirm机制是异步的,你发送个消息之后就可以发送下一个消息,然后那个消息RabbitMQ接收了之后会异步回调你一个接口通知你这个消息接收到了。

  所以一般在生产者这块避免数据丢失,都是用confirm机制的。

(2)RabbitMQ弄丢了数据

  就是RabbitMQ自己弄丢了数据,这个你必须开启RabbitMQ的持久化,就是消息写入之后会持久化到磁盘,哪怕是RabbitMQ自己挂了,恢复之后会自动读取之前存储的数据,一般数据不会丢。除非极其罕见的是,RabbitMQ还没持久化,自己就挂了,可能导致少量数据会丢失的,但是这个概率较小。

设置持久化有两个步骤

第一个是创建queue的时候将其设置为持久化的,这样就可以保证RabbitMQ持久化queue的元数据,但是不会持久化queue里的数据;

第二个是发送消息的时候将消息的deliveryMode设置为2,就是将消息设置为持久化的,此时RabbitMQ就会将消息持久化到磁盘上去。必须要同时设置这两个持久化才行,RabbitMQ哪怕是挂了,再次重启,也会从磁盘上重启恢复queue,恢复这个queue里的数据。

而且持久化可以跟生产者那边的confirm机制配合起来,只有消息被持久化到磁盘之后,才会通知生产者ack了,所以哪怕是在持久化到磁盘之前,RabbitMQ挂了,数据丢了,生产者收不到ack,你也是可以自己重发的。

哪怕是你给RabbitMQ开启了持久化机制,也有一种可能,就是这个消息写到了RabbitMQ中,但是还没来得及持久化到磁盘上,结果不巧,此时RabbitMQ挂了,就会导致内存里的一点点数据会丢失。

(3)消费端弄丢了数据

  RabbitMQ如果丢失了数据,主要是因为你消费的时候,刚消费到,还没处理,结果进程挂了,比如重启了,那么就尴尬了,RabbitMQ认为你都消费了,这数据就丢了。

  这个时候得用RabbitMQ提供的ack机制,简单来说,就是你关闭RabbitMQ自动ack,可以通过一个api来调用就行,然后每次你自己代码里确保处理完的时候,在程序里ack一把。这样的话,如果你还没处理完,不就没有ack?那RabbitMQ就认为你还没处理完,这个时候RabbitMQ会把这个消费分配给别的consumer去处理,消息是不会丢的。

在发送消息的时候,接受时记录 DB 日志,定时轮询 DB 日志,查明那些发送的消息没有成功消费,启动重新发送消息机制。

消息顺序

场景: 比如下单操作,下单成功后,会发布创建订单和减库存的消息,但扣库存消息执行会先于创建订单的消息,也就说前者执行成功之后,才能执行后者。

MQ 层面支持消息的顺序处理开销太大了,为了极少量的需求,增加了整体上的复杂性。应该尽可能的在应用层面进行处理。

解决方式如下:

  1. 同步执行,当一个消息执行之后,再发送下一个消息。
  2. rabbitmq:拆分多个queue,每个queue一个consumer,就是多一些queue而已,确实是麻烦点;或者就一个queue但是对应一个consumer,然后这个consumer内部用内存队列做排队,然后分发给底层不同的worker来处理。

消息重复

如果消费端接受到两个一样的消息,应该如何处理呢?

  1. 消费端处理消息的业务逻辑应该保持幂等性。
  2. 保证每条消息都有唯一标号,首先检查执行成功的日志中是否存有该消息的 ID 如果没有则执行,如果已经存在则丢弃消息。如果在消息系统中实现,会对消息系统的吞吐量造成影响。所以还是在业务端进行去重处理。

事务消息

具体来说,就是把消息的发送分成了2个阶段:Prepare阶段和确认阶段。

具体来说,上面的2个步骤,被分解成3个步骤:
(1) 发送Prepared消息
(2) update DB
(3) 根据update DB结果成功或失败,Confirm 或者取消 Prepared 消息。

可能有人会问了,前2步执行成功了,最后1步失败了怎么办?这里就涉及到了RocketMQ的关键点:RabbitMQ 会定期(默认是1分钟)扫描所有的Prepared消息,询问发送方,到底是要确认这条消息发出去?还是取消此条消息?

消息阻塞

上千万条消息在 mq 里积压了几个小时了还没解决

紧急扩容

1)先修复 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机器来消费消息

消息队列过期失效问题

 假设你用的是rabbitmq,rabbitmq是可以设置过期时间的,就是TTL,如果消息在queue中积压超过一定的时间就会被rabbitmq给清理掉,这个数据就没了。晚上12点以后,用户都睡觉了。

  这个时候我们就开始写程序,将丢失的那批数据,写个临时程序,一点一点的查出来,然后重新灌入mq里面去,把白天丢的数据给他补回来。也只能是这样了。

  假设1万个订单积压在mq里面,没有处理,其中1000个订单都丢了,你只能手动写程序把那1000个订单给查出来,手动发到mq里去再补一次。

消息队列满

丢弃,晚上补数据

-------------本文结束感谢您的阅读-------------