RabbitMQ进阶

转载：https://blog.csdn.net/feiying0canglang/article/details/105351740

消息丢失

其他网址

官方文档：https://www.rabbitmq.com/confirms.html

【消息队列】RabbitMQ如何处理消息丢失 - 个人文章 - SegmentFault 思否
RabbitMQ防止消息丢失_大数据_杨航的专栏-CSDN博客

简介

首先明确一点 一条消息的传送流程：生产者->MQ->消费者

所以有三个地方都会丢失数据：

1. Producer端：发送消息过程中出现网络问题：producer以为发送成功，但RabbitMQ server没有收到；
2. RabbitMQ server 端：接收到消息后由于服务器宕机或重启等原因（消息默认存在内存中）导致消息丢失；
3. Consumer端：Consumer端接收到消息后自动返回ack，但后边处理消息出错，没有完成消息的处理；

生产者弄丢了数据

什么时候会丢数据？

生产者将数据发送到RabbitMQ的时候，可能因为网络问题导致数据没到达RabbitMQ Server。

解决方案1.发送回执确认（推荐，最常用）

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

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

解决方案2：本地消息表+定时任务（推荐）

见：分布式事务系列--消息表+MQ_feiying0canglang的博客-CSDN博客
            =>本地消息表+MQ（无事务）=> 实例流程

解决方案3：使用事务（性能差）

        可以选择用RabbitMQ提供的事务功能，在生产者发送数据之前开启RabbitMQ事务（channel.txSelect），然后发送消息，如果消息没有成功被RabbitMQ接收到，那么生产者会收到异常报错，此时就可以回滚事务（channel.txRollback），然后重试发送消息；如果收到了消息，那么可以提交事务（channel.txCommit）。但是问题是，开始RabbitMQ事务机制，基本上吞吐量会下来，因为太耗性能。

RabbitMQ服务器弄丢了数据

什么时候会丢数据？

接收到消息后由于服务器宕机或重启等原因（消息默认存在内存中）导致消息丢失；

解决方案：开启RabbitMQ的数据持久化

结论

为了防止RabbitMQ自己弄丢了数据，这个你必须开启RabbitMQ的持久化，就是消息写入之后会持久化到磁盘，哪怕是RabbitMQ自己挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。

详解

　　设置持久化有两个步骤，第一个是创建queue的时候将其设置为持久化的，这样就可以保证RabbitMQ持久化queue的元数据，但是不会持久化queue里的数据；第二个是发送消息的时候将消息的deliveryMode设置为2，就是将消息设置为持久化的，此时RabbitMQ就会将消息持久化到磁盘上去。必须要同时设置这两个持久化才行，RabbitMQ哪怕是挂了，再次重启，也会从磁盘上重启恢复queue，恢复这个queue里的数据。

　　极其罕见的是，RabbitMQ还没持久化，自己就挂了，导致数据丢失，但是这个概率较小。当然，也可以解决，解决方法如下：

　　RabbitMQ开启持久化，生产者开启confirm机制。只有消息被持久化到磁盘之后，才会通知生产者ack了，所以哪怕是在持久化到磁盘之前，RabbitMQ挂了，数据丢了，生产者收不到ack，你也是可以自己重发的。

消费端弄丢了数据

什么时候会丢数据？

Consumer端接收到消息后自动返回ack，但后边处理消息出错，没有完成消息的处理；

解决方案

结论

设置返回确认的模式为手动，并在处理完消息后手动去提交确认。

详解

这样配置之后，如果MQ等待一段时间后你没有发送过来处理完成的回执，那么RabbitMQ就认为你还没处理完，这个时候RabbitMQ会把这个消费分配给别的consumer去处理，消息是不会丢的。

重复消费

什么时候会重复消费

1.自动提交模式

消费者受到消息后，要自动提交，但提交后，网络出故障，RabbitMQ服务器没收到提交消息，那么此消息会被重新放入队列，会再次发给消费者。

2.手动提交模式

（1）网络故障问题，同上

（2）接收到消息并处理结束了，此时消费者挂了，没有手动提交消息。

解决方案

简介

造成消息重复的两个原因是：网络不可达、消费端宕机。无法避免这两个问题。解决这个问题的办法就是绕过这个问题。

问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？有以下两个方法

1. 消费端处理消息的业务逻辑保持幂等性。
2. 保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现。

项	第一种	第二种
说明	只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样	利用一张日志表来记录已经处理成功的消息的 ID，如果新到的消息 ID 已经在日志表中，那么就不再处理这条消息。
实现者	应该在消费端实现，不属于消息系统要实现的功能。	可以MQ服务器实现，也可以消费端实现。正常情况下出现重复消息的概率其实很小，如果由消息系统来实现的话，肯定会对消息系统的吞吐量和高可用有影响，所以最好还是由消费端自己处理消息重复的问题，这也是 RabbitMQ 不解决消息重复的问题的原因。

第一种方法

• 比如你拿个数据要写库，你先根据主键查一下，如果这数据都有了，你就别插入了，update 一下好吧。
• 比如你是写 Redis，那没问题了，反正每次都是 set，天然幂等性。
• 比如你不是上面两个场景，那做的稍微复杂一点，你需要让生产者发送每条数据的时候，里面加一个全局唯一的 id，类似订单 id 之类的东西，然后你这里消费到了之后，先根据这个 id 去比如 Redis 里查一下，之前消费过吗？如果没有消费过，你就处理，然后这个 id 写 Redis。如果消费过了，那你就别处理了，保证别重复处理相同的消息即可。

实例网址:RocketMQ重复消息终极解决方案 - 简书

第二种方法

【RabbitMQ】保证消息的不重复消费_大数据_千千-CSDN博客

顺序性

其他网址

关于MQ的几件小事（五）如何保证消息按顺序执行 - 简书
聊一聊顺序消息（RocketMQ顺序消息的实现机制） - 杭州.Mark - 博客园

消息堆积

消息堆积产生原因

消息堆积即消息没及时被消费，是生产者生产消息速度快于消费者消费的速度导致的，消费者消费慢可能是因为：本身逻辑耗费时间较长、阻塞了

预防措施

生产者

1. 给消息设置年龄，超时就丢弃
2. 考虑使用队列最大长度限制
3. 减少发布频率

消费者

1. 增加消费者的处理能力   //优化代码；使用JDK的队列缓存数据，多线程去处理（一般考虑顺序问题，采用单例线程）
2. 建立新的queue，消费者同时订阅新旧queue，采用订阅模式
3. 默认情况下，rabbitmq消费者为单线程串行消费（org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer类的concurrentConsumers与txSize（对应prefetchCount）都是1），设置并发消费两个关键属性concurrentConsumers和prefetchCount。concurrentConsumers：设置的是对每个listener在初始化的时候设置的并发消费者的个数；prefetchCount：每次从broker里面取的待消费的消息的个数。
   配置方法：修改application.properties：

   spring.rabbitmq.listener.concurrency=m
   spring.rabbitmq.listener.prefetch=n

   Spring Amqp的解释：

   prefetchCount(prefetch)
       The number of messages to accept from the broker in one socket frame. The higher this is the faster the messages can be delivered, but the higher the risk of non-sequential processing. Ignored if the acknowledgeMode
       is NONE. This will be increased, if necessary, to match the txSize
   
   concurrentConsumers(concurrency)
       The number of concurrent consumers to initially start for each listener.

   其他网址：RabbitMQ消费者的几个参数 - 简书

综合（使用缓存）：

• 生产者端缓存数据，在mq被消费完后再发送到mq，打破发送循环条件。设置合适的qos值（channel.BasicQos()方法：每次从队列拉取的消息数量），当qos值被用光，而新的ack没有被mq接收时，就可以跳出发送循环，去接收新的消息。
• 消费者主动block接收进程，消费者感受到接收消息过快时主动block，利用block和unblock方法调节接收速率，当接收线程被block时，跳出发送循环。

已出事故的解决措施

其他网址：完了！生产事故！几百万消息在消息队列里积压了几个小时！_一诺-CSDN博客_消息队列消费者来不及消耗

情况1：堆积的消息还需要使用

方案1：简单修复

修复consumer的问题，让他恢复消费速度，然后等待几个小时消费完毕

方案2：复杂修复

临时紧急扩容了，具体操作步骤和思路如下：

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机器来消费消息

情况2：堆积的消息不需要使用

删除消息即可。

另见：RabbitMQ系列--使用_feiying0canglang的博客-CSDN博客