RabbitMq(七) -- 常见问题：幂等性问题（消息重复消费）、消息丢失_rabbitmq重复消费问题_2024-04-20 17:28:27.532 error 61884 --- [nio-9090-

先自我介绍一下，小编浙江大学毕业，去过华为、字节跳动等大厂，目前阿里P7

深知大多数程序员，想要提升技能，往往是自己摸索成长，但自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年最新大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！

由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新

如果你需要这些资料，可以添加V获取：vip204888 （备注大数据）

正文

private RabbitTemplate rabbitTemplate;

//依赖注入 rabbitTemplate 之后再设置它的回调对象
// 此注解会在其他注解执行完成后再执行，所以rabbitTemplate先注入，再执行此初始化方法
@PostConstruct
public void init() {
    // 设置rabbitTemplate的ConfirmCallBack为我们重写后的类
    rabbitTemplate.setConfirmCallback(this);
    rabbitTemplate.setReturnCallback(this);
}
/\*\*
1
2
3
4
5
6
7
8
9
10
11

* 交换机不管是否收到消息都会执行的一个回调方法
*
* @param correlationData 消息相关数据
* @param ack 交换机是否收到消息
* @param cause 未收到消息的原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : “”;
if (ack) {
log.info(“交换机已经收到 id 为:{}的消息”, id);
} else {
log.info(“交换机还未收到 id 为:{}消息，原因:{}”, id, cause);
}
}

// 确认消息是否从交换机成功到达队列中，失败将会执行，成功则不执行
@Override
public void returnedMessage(Message message, int replayCode, String replayText, String exchange, String routingKey) {
    log.info("消息{}，被交换机{}退回，退回原因：{}，路由key：", new String(message.getBody()), exchange, replayText, routingKey);
}
1
2
3
4
5

}

###### 1.4.4 数据库对象相关配置：


数据库脚本：



1
2
3
4
5
6
7
8

CREATE TABLE message\_idempotent (
message\_id varchar(50) NOT NULL COMMENT ‘消息ID’,
message\_content varchar(2000) DEFAULT NULL COMMENT ‘消息内容’,
PRIMARY KEY (message\_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

对象：



1
2
3
4
5

@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageIdempotent extends Model {

@TableId("message\_id")
private String messageId;

@TableField("message\_content")
private String messageContent;
1
2
3
4
5

}

mapper：



1
2
3
4
5

@Mapper
public interface MessageIdempotentMapper extends BaseMapper {
}

###### 1.4.5 生产者编写：



1
2
3
4
5

/**
* 消息幂等性
* */
@GetMapping(“/sendMessage”)
public void sendMessage(String msg, String id) {
MessageProperties messageProperties = new MessageProperties();
messageProperties.setMessageId(id);
messageProperties.setContentType(“text/plain”);
messageProperties.setContentEncoding(“utf-8”);
Message message = new Message(msg.getBytes(), messageProperties);
log.info(“生产消息:” + message.toString());
// 消息发送确认回调
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(“DirectExchange-01”, “DirectRouting01”, message, correlationData);
}

访问接口：



1
2
3
4
5

http://localhost:8091/shiro/revisit/sendMessage?msg=你好啊&id=1
http://localhost:8091/shiro/revisit/sendMessage?msg=&id=1

日志：（此处有confirmCallback未回调问题待解决，按道理打印完生产消息后应该打印：交换机已经收到 id 为:{}的消息）



1
2
3
4
5

2023-04-10 14:31:12.859 INFO 19232 — [nio-8091-exec-1] c.y.t.r.TestRevisit.RevisitController : 生产消息:(Body:‘你好啊’ MessageProperties [headers={}, messageId=1, contentType=text/plain, contentEncoding=utf-8, contentLength=0, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])
2023-04-10 14:31:29.002 INFO 19232 — [nio-8091-exec-2] c.y.t.r.TestRevisit.RevisitController : 生产消息:(Body:‘’ MessageProperties [headers={}, messageId=1, contentType=text/plain, contentEncoding=utf-8, contentLength=0, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])

客户端中：  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/1ecdbc710e4c47be84e7fede4904b3a5.png)


###### 1.4.6 消费者编写：



1
2
3
4
5
6
7
8
9

@RabbitListener(queues = “DirectQueue-01”)
public void receiveMessage02(Message message, Channel channel) throws IOException {
String messageId = message.getMessageProperties().getMessageId();
String messageContent = new String(message.getBody(), StandardCharsets.UTF_8);

MessageIdempotent messageIdempotent = new MessageIdempotent();
messageIdempotent.setMessageId(messageId);
messageIdempotent.setMessageContent(messageContent);

try {
    if (messageIdempotentMapper.insert(messageIdempotent) <= 0) {
        log.info("DirectQueue-01-消费者收到消息，消息ID：" + messageId + " 消息内容：" + messageContent);
        // 消息确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } else {
        log.info("消息 " + messageId + " 已经消费过！");
    }
} catch (Exception e) {
    log.info("消息 " + messageId + " 已经消费过！");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

}

结果：



1
2
3
4
5

2023-04-10 14:47:06.738 INFO 25416 — [ntContainer#6-1] c.y.t.r.TestRevisit.RevisitConsumer : DirectQueue-01-消费者收到消息，消息ID：1 消息内容：你好啊
2023-04-10 14:47:06.745 INFO 25416 — [ntContainer#6-1] c.y.t.r.TestRevisit.RevisitConsumer : 消息 1 已经消费过！

数据库中：  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/dec70cabbf53414fbd39dd942a02a3ba.png)  
 队列中：  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/9e5d89eeba18488e8561664ee257481b.png)


#### 1.5 note Redis 原子性


利用 redis 执行 setnx 命令，天然具有幂等性，从而实现不重复消费。利用redis的操作的好处是缓存更快。


代码这里不再演示，无非是一个插入数据库，一个setnx进redis。


## 2. 消息丢失


#### 2.1 消息丢失的场景


![在这里插入图片描述](https://img-blog.csdnimg.cn/894b66b0060a4fb5960266bd1545d1fd.png)


* 第一种：生产者弄丢了数据。生产者将数据发送到 RabbitMQ 的时候，可能数据就在半路给搞丢了，因为网络问题啥的，都有可能。
* 第二种：RabbitMQ 弄丢了数据。MQ还没有持久化自己挂了
* 第三种：消费端弄丢了数据。刚消费到，还没处理，结果进程挂了，比如重启了。


#### 2.2 RabbitMQ消息丢失解决方案


![在这里插入图片描述](https://img-blog.csdnimg.cn/91dca65b161d401cb2cb26e4efda2e4e.png)


##### 2.2.1 针对生产者


###### 1. 方案1 ：开启RabbitMQ事务


可以选择用 RabbitMQ 提供的事务功能，就是生产者发送数据之前开启 RabbitMQ 事务`channel.txSelect`，然后发送消息，如果消息没有成功被 RabbitMQ 接收到，那么生产者会收到异常报错，此时就可以回滚事务`channel.txRollback`，然后重试发送消息；如果收到了消息，那么可以提交事务channel.txCommit。



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// 开启事务
channel.txSelect
try {
// 这里发送消息
} catch (Exception e) {
channel.txRollback

// 这里再次重发这条消息

}
// 提交事务
channel.txCommit

**缺点**：  
 RabbitMQ 事务机制是同步的，你提交一个事务之后会阻塞在那儿，采用这种方式基本上吞吐量会下来，因为太耗性能。


###### 2. 方案2： 使用confirm机制


事务机制和 confirm 机制最大的不同在于，事务机制是同步的，你提交一个事务之后会阻塞在那儿，但是 confirm 机制是异步的


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


即第一节MyCallback中：



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

/**
* 交换机不管是否收到消息都会执行的一个回调方法
*
* @param correlationData 消息相关数据
* @param ack 交换机是否收到消息
* @param cause 未收到消息的原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : “”;
if (ack) {
log.info(“交换机已经收到 id 为:{}的消息”, id);
} else {
log.info(“交换机还未收到 id 为:{}消息，原因:{}”, id, cause);
}
}

##### 2.2.2 针对RabbitMQ


说三点：


1. 要保证rabbitMQ不丢失消息，那么就需要开启rabbitMQ的持久化机制，即把消息持久化到硬盘上，这样即使rabbitMQ挂掉在重启后仍然可以从硬盘读取消息；
2. 如果rabbitMQ单点故障怎么办，这种情况倒不会造成消息丢失，这里就要提到rabbitMQ的3种安装模式，单机模式、普通集群模式、镜像集群模式，这里要保证rabbitMQ的高可用就要配合HAPROXY做镜像集群模式
3. 如果硬盘坏掉怎么保证消息不丢失


###### 1. 消息持久化


RabbitMQ 的消息默认存放在内存上面，如果不特别声明设置，消息不会持久化保存到硬盘上面的，如果节点重启或者意外crash掉，消息就会丢失。


所以就要对消息进行持久化处理。如何持久化，下面具体说明下：


要想做到消息持久化，必须满足以下三个条件，缺一不可。


1. Exchange 设置持久化
2. Queue 设置持久化
3. Message持久化发送：发送消息设置发送模式deliveryMode=2，代表持久化消息


###### 2. 设置集群镜像模式


我们先来介绍下RabbitMQ三种部署模式：


1. 单节点模式：最简单的情况，非集群模式，节点挂了，消息就不能用了。业务可能瘫痪，只能等待。
2. 普通模式：消息只会存在与当前节点中，并不会同步到其他节点，当前节点宕机，有影响的业务会瘫痪，只能等待节点恢复重启可用（必须持久化消息情况下）。
3. 镜像模式：消息会同步到其他节点上，可以设置同步的节点个数，但吞吐量会下降。属于RabbitMQ的HA方案


为什么设置镜像模式集群，因为队列的内容仅仅存在某一个节点上面，不会存在所有节点上面，所有节点仅仅存放消息结构和元数据。下面自己画了一张图介绍普通集群丢失消息情况：  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/3aed3e80101c4c76b3a323864e08527b.png)  
 如果想解决上面途中问题，保证消息不丢失，需要采用HA 镜像模式队列。


下面介绍下三种HA策略模式：


1. 同步至所有的
2. 同步最多N个机器
3. 只同步至符合指定名称的nodes


命令处理HA策略模版：`rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]`


1. 为每个以“rock.wechat”开头的队列设置所有节点的镜像，并且设置为自动同步模式



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

rabbitmqctl set_policy ha-all “^rock.wechat” ‘{“ha-mode”:“all”,“ha-sync-mode”:“automatic”}’

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化的资料的朋友，可以添加V获取：vip204888 （备注大数据）

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

rabbitmqctl set_policy ha-all “^rock.wechat” ‘{“ha-mode”:“all”,“ha-sync-mode”:“automatic”}’

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化的资料的朋友，可以添加V获取：vip204888 （备注大数据）
[外链图片转存中…(img-9DRPR8Bv-1713320069580)]

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！