MQ实际场景解决方案参考

1、MQ持久化机制

如果我们生产者每发一条消息&＃xff0c;都要MQ持久化到磁盘中&＃xff0c;然后再发起ack或nack的回调。这样的话是不是我们MQ的吞吐量很不高&＃xff0c;因为每次都要把消息持久化到磁盘中。写入磁盘这个动作是很慢的&＃xff0c;在高并发场景下是不能够接受的&＃xff0c;吞吐量太低了。

所以MQ持久化磁盘真实的实现&＃xff0c;是通过异步调用处理的&＃xff0c;他是有一定的机制&＃xff0c;如&＃xff1a;等到有几千条消息的时候&＃xff0c;会一次性的刷盘到磁盘上面。而不是每来一条消息&＃xff0c;就刷盘一次。

2、confirm机制的原理

&＃xff08;1&＃xff09;消息生产者把消息发送给MQ&＃xff0c;如果接收成功&＃xff0c;MQ会返回一个ack消息给生产者&＃xff1b;

&＃xff08;2&＃xff09;如果消息接收不成功&＃xff0c;MQ会返回一个nack消息给生产者&＃xff1b;

3、消息幂等

幂等性&＃xff1a;就是用户对于同一操作发起的一次请求或者多次请求的结果是一致的&＃xff0c;不会因为多次点击而产生了副作用&＃xff0c;数据库的结果都是唯一的&＃xff0c;不可变的。
只要保持幂等性&＃xff0c;不管来多少条重复消息&＃xff0c;最后处理的结果都一样&＃xff0c;需要业务端来实现。 

4、消息堆积

消息中间件的主要功能是异步解耦&＃xff0c;还有个重要功能是挡住前端的数据洪峰&＃xff0c;保证后端系统的稳定性&＃xff0c;这就要求消息中间件具有一定的消息堆积能力&＃xff0c;消息堆积分以下两种情况&＃xff1a; 

1. 消息堆积在内存Buffer&＃xff1a;一旦超过内存Buffer&＃xff0c;可以根据一定的丢弃策略来丢弃消息&＃xff0c;如CORBA Notification规范中描述。适合能容忍丢弃消息的业务&＃xff0c;这种情况消息的堆积能力主要在于内存Buffer大小&＃xff0c;而且消息堆积后&＃xff0c;性能下降不会太大&＃xff0c;因为内存中数据多少对于对外提供的访问能力影响有限。
2. 消息堆积到持久化存储系统中&＃xff1a;例如DB&＃xff0c;KV存储&＃xff0c;文件记录形式。 当消息不能在内存Cache命中时&＃xff0c;要不可避免的访问磁盘&＃xff0c;会产生大量读IO&＃xff0c;读IO的吞吐量直接决定了消息堆积后的访问能力。

1、丢失原因&＃xff1f;

第一种情况&＃xff1a;一般MQ中间件为了提高系统的吞吐量会把消息保存在内存中&＃xff0c;如果不作其他处理&＃xff0c;MQ服务器一旦宕机&＃xff0c;消息将全部丢失。这个是业务不允许的&＃xff0c;造成很大的影响。

第二种情况&＃xff1a; 就是消息刚刚保存到MQ内存中&＃xff0c;但还没有来得及更新到磁盘文件中&＃xff0c;突然宕机了。这个场景在持续的大量消息投递的过程中&＃xff0c;会很常见。

2、避免消息丢失

持久化&＃xff1a;将MQ消息持久化&＃xff1b;

confirm机制&＃xff1a;消息发送端接受MQ的confirm通知&＃xff1b;

通过持久化和confirm机制&＃xff0c;这样是不是就可以保障100%消息不丢失了呢&＃xff1f;

comfirm机制其实是一个异步监听的机制&＃xff0c;是为了保证系统的高吞吐量&＃xff0c;这样就导致了还是不能够100%保障消息不丢失&＃xff0c;因为即使加上了confirm机制&＃xff0c;消息在MQ内存中还没有刷盘到磁盘就宕机了&＃xff0c;还是没法处理。

3、辅助手段

生产者在投递消息之前&＃xff0c;可以在本地数据库建一张消息表&＃xff0c;先把消息持久化到Redis或DB中&＃xff0c;这样就可以利用本地数据库的事务机制。事务提交成功后&＃xff0c;将消息表中的消息转移到消息队列中。

confirm机制监听消息是否发送成功&＃xff1f;如ack成功消息&＃xff0c;删除DB中此消息或者修改消息表状态。

如果nack不成功的消息&＃xff0c;这个可以根据自身的业务选择是否重发此消息。也可以删除此消息&＃xff0c;由自己的业务决定。

1、重复消费的问题&＃xff1f;

导致重复消费的原因可能出现在生产者&＃xff0c;也可能出现在 MQ 或 消费者。

这里说的重复消费问题是指同一个数据被执行了两次&＃xff0c;不单单指 MQ 中一条消息被消费了两次&＃xff0c;也可能是 MQ 中存在两条一模一样的消费。

• 生产者&＃xff1a;生产者可能会重复推送一条数据到 MQ 中&＃xff0c;为什么会出现这种情况呢&＃xff1f;也许是一个 Controller 接口被重复调用了 2 次&＃xff0c;没有做接口幂等性导致的&＃xff1b;也可能是推送消息到 MQ 时响应比较慢&＃xff0c;生产者的重试机制导致再次推送了一次消息。
• MQ&＃xff1a;在消费者消费完一条数据响应 ack 信号消费成功时&＃xff0c;MQ 突然挂了&＃xff0c;导致 MQ 以为消费者还未消费该条数据&＃xff0c;MQ 恢复后再次推送了该条消息&＃xff0c;导致了重复消费。
• 消费者&＃xff1a;消费者已经消费完了一条消息&＃xff0c;正准备但是还未给 MQ 发送 ack 信号时&＃xff0c;此时消费者挂了&＃xff0c;服务重启后 MQ 以为消费者还没有消费该消息&＃xff0c;再次推送了该条消息。

2、如何保证幂等性&＃xff1f;

去重原则&＃xff1a;使用业务端逻辑保持幂等性

去重策略&＃xff1a;保证每条消息都有唯一编号(比如唯一流水号)&＃xff0c;且保证消息处理成功与去重表的日志同时出现。
建立一个消息表&＃xff0c;拿到这个消息做数据库的insert操作。给这个消息做一个唯一主键&＃xff08;primary key&＃xff09;或者唯一约束&＃xff0c;那么就算出现重复消费的情况&＃xff0c;就会导致主键冲突&＃xff0c;那么就不再处理这条消息。

消费者怎么解决重复消费问题呢&＃xff1f;这里提供两种方法&＃xff1a;

1. 状态判断法&＃xff1a;消费者消费数据后把消费数据记录在 redis 中&＃xff0c;下次消费时先到 redis 中查看是否存在该消息&＃xff0c;存在则表示消息已经消费过&＃xff0c;直接丢弃消息。
2. 业务判断法&＃xff1a;通常数据消费后都需要插入到数据库中&＃xff0c;使用数据库的唯一性约束防止重复消费。每次消费直接尝试插入数据&＃xff0c;如果提示唯一性字段重复&＃xff0c;则直接丢失消息。一般都是通过这个业务判断的方法就可以简单高效地避免消息的重复处理了。

1、消息积压处理办法

首要考虑临时紧急扩容&＃xff0c;比如&＃xff1a;利用K8s弹性扩容&＃xff0c;增加消费者数量&＃xff1b;

先修复 consumer 的问题&＃xff0c;确保其恢复消费速度&＃xff0c;然后将现有 cnosumer 都停掉。

新建一个 topic&＃xff0c;partition 是原来的 10 倍&＃xff0c;临时建立好原先 10 倍的 queue 数量。

然后写一个临时的分发数据的 consumer 程序&＃xff0c;这个程序部署上去消费积压的数据&＃xff0c;消费之后不做耗时的处理&＃xff0c;直接均匀轮询写入临时建立好的 10 倍数量的 queue。

接着临时征用 10 倍的机器来部署 consumer&＃xff0c;每一批 consumer 消费一个临时 queue 的数据。这种做法相当于是临时将 queue 资源和consumer 资源扩大 10 倍&＃xff0c;以正常的 10 倍速度来消费数据。

等快速消费完积压数据之后&＃xff0c;得恢复原先部署的架构&＃xff0c;重新用原先的consumer 机器来消费消息。

2、MQ中消息失效

假设你用的是 RabbitMQ&＃xff0c;RabbtiMQ 是可以设置过期时间的&＃xff0c;也就是 TTL。如果消息在queue 中积压超过一定的时间就会被 RabbitMQ 给清理掉&＃xff0c;这个数据就没了。那这就是第二个坑了。这就不是说数据会大量积压在 mq里&＃xff0c;而是大量的数据会直接搞丢。

我们可以采取一个方案&＃xff0c;就是批量重导&＃xff0c;这个我们之前线上也有类似的场景干过。就是大量积压的时候&＃xff0c;我们当时就直接丢弃数据了&＃xff0c;然后等过了高峰期以后&＃xff0c;比如&＃xff1a;

大家一起喝咖啡熬夜到晚上12点以后&＃xff0c;用户都睡觉了。这个时候我们就开始写程序&＃xff0c;将丢失的那批数据&＃xff0c;写个临时程序&＃xff0c;一点一点的查出来&＃xff0c;然后重新灌入 mq 里面去&＃xff0c;把白天丢的数据给他补回来。也只能是这样了。

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

3、MQ消息队列快满了

如果消息积压在 mq 里&＃xff0c;你很长时间都没有处理掉&＃xff0c;此时导致 mq 都快写满了&＃xff0c;咋办&＃xff1f;这个还有别的办法吗&＃xff1f;

没有&＃xff0c;谁让你第一个方案执行的太慢了&＃xff0c;你临时写程序&＃xff0c;接入数据来消费&＃xff0c;消费一个丢弃一个&＃xff0c;都不要了&＃xff0c;快速消费掉所有的消息。然后走第二个方案&＃xff0c;到了晚上再补数据吧 。

1、为什么要保证顺序 

消息队列中的若干消息如果是对同一个数据进行操作&＃xff0c;这些操作具有前后的关系&＃xff0c;必须要按前后的顺序执行&＃xff0c;否则就会造成数据异常。举例&＃xff1a;

Producer先后发送了2条消息&＃xff0c;一条insert&＃xff0c;一条update&＃xff0c;分别分配到2台Queue中&＃xff0c;消费者组中的两台机器分别处理这两个Queue中的消息&＃xff0c;这时候顺序是无法保证的&＃xff1b; 

2、出现顺序错乱的场景 

&＃xff08;1&＃xff09;RabbtiMQ

• a、一个queue&＃xff0c;有多个consumer去消费&＃xff0c;这样就会造成顺序的错误。consumer从MQ里面读取数据是有序的&＃xff0c;但是每个consumer的执行时间是不固定的&＃xff0c;无法保证先读到消息的consumer一定先完成操作&＃xff0c;这样就会出现消息并没有按照顺序执行&＃xff0c;造成数据顺序错误。
• b、一个queue对应一个consumer&＃xff0c;但是consumer里面进行了多线程消费&＃xff0c;这样也会造成消息消费顺序错误。 

&＃xff08;2&＃xff09;RocketMQ

• a、RocketMQ一个topic&＃xff0c;一个queue&＃xff0c;一个consumer&＃xff0c;但是consumer内部进行多线程消费&＃xff0c;这样数据也会出现顺序错乱问题。
• b、具有顺序的数据写入到了不同的partition里面&＃xff0c;不同的消费者去消费&＃xff0c;但是每个consumer的执行时间是不固定的&＃xff0c;无法保证先读到消息的consumer一定先完成操作&＃xff0c;这样就会出现消息并没有按照顺序执行&＃xff0c;造成数据顺序错误。

3、保证消息的消费顺序 

&＃xff08;1&＃xff09;RabbtiMQ

• a、拆分多个queue&＃xff0c;每个queue一个consumer&＃xff0c;就是多一些queue而已&＃xff0c;确实是麻烦点&＃xff1b;这样也会造成吞吐量下降&＃xff0c;可以在消费者内部采用多线程的方式取消费。
• b、或者就一个queue但是对应一个consumer&＃xff0c;然后这个consumer内部用内存队列做排队&＃xff0c;然后分发给底层不同的worker来处理。

&＃xff08;2&＃xff09;RocketMQ/Kafka

RocketMQ可以严格的保证消息有序。但这个顺序&＃xff0c;不是全局顺序&＃xff0c;只是分区&＃xff08;queue&＃xff09;顺序。要全局顺序只能一个分区。 之所以出现你这个场景看起来不是顺序的&＃xff0c;是因为发送消息的时候&＃xff0c;消息发送默认是会采用轮询的方式发送到不通的queue&＃xff08;分区&＃xff09;。

• a、分区顺序&＃xff1a;拆分多个queue&＃xff0c;然后N个线程分别消费一个queue。
• b、全局顺序&＃xff1a;确保同一个消息发送到同一个partition&＃xff0c;一个topic&＃xff0c;一个分区&＃xff08;queue&＃xff09;&＃xff0c;一个consumer&＃xff0c;内部单线程消费。 

启动一个cron定时任务&＃xff0c;每隔一段时间执行一次&＃xff0c;比如30分钟&＃xff0c;找到那些超时的数据&＃xff0c;直接更新状态&＃xff0c;或者拿出来执行一些操作。如果数据量比较大&＃xff0c;需要分页查询&＃xff0c;分页update&＃xff0c;这将是一个for循环更新操作。

cron方案是很常见的一种方案&＃xff0c;但是常见的不一定是最好的&＃xff0c;主要有以下几个问题&＃xff1a;

• 当数据量大的时候轮询效率低&＃xff1b;
• 时效性不够好&＃xff0c;如果每小时轮询一次&＃xff0c;最差的情况时间误差会达到1小时&＃xff1b;
• 如果通过增加cron轮询频率来减少时间误差&＃xff0c;则会出现轮询低效和重复计算的问题&＃xff1b;

&＃xff08;1&＃xff09;RabbtiMQ

RabbitMQ延迟队列的核心思路是&＃xff1a;TTL消息/队列 &＃43; DLX死信队列

TTL是Time To Live的缩写&＃xff0c;也就是生存时间的意思&＃xff1b;

DLX死信队列虽然叫队列&＃xff0c;但其实指的是Exchange&＃xff0c;或者说指的Exchange和它所属的Queue&＃xff0c;他俩一块构成了死信队列。

&＃xff08;2&＃xff09;RocketMQ

RocketMQ延迟队列的核心思路是&＃xff1a;所有的延迟消息由producer发出之后&＃xff0c;都会存放到同一个topic&＃xff08;SCHEDULE_TOPIC_XXXX&＃xff09;下&＃xff0c;不同的延迟级别会对应不同的队列序号&＃xff0c;当延迟时间到之后&＃xff0c;由定时线程读取转换为普通的消息存的真实指定的topic下&＃xff0c;此时对于consumer端此消息才可见&＃xff0c;从而被consumer消费。

注意&＃xff1a;RocketMQ不支持任意时间的延时&＃xff0c;只支持以下几个固定的延时等级
private String messageDelayLevel &＃61; "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

回溯消费是指Consumer已经消费成功的消息&＃xff0c;由于业务上的需求需要重新消费&＃xff0c;要支持此功能&＃xff0c;Broker在向Consumer投递成功消息后&＃xff0c;消息仍然需要保留。并且重新消费一般是按照时间维度。

例如&＃xff1a;由于Consumer系统故障&＃xff0c;恢复后需要重新消费1小时前的数据&＃xff0c;那么Broker要提供一种机制&＃xff0c;可以按照时间维度来回退消费进度。

RocketMQ支持按照时间回溯消费&＃xff0c;时间维度精确到毫秒&＃xff0c;可以向前回溯&＃xff0c;也可以向后回溯。

感觉从生产端回溯更加合理&＃xff0c;需要生产端提供可回溯的业务机制。