RabbitMq(七) -- 常见问题：幂等性问题（消息重复消费）、消息丢失_rabbitmq重复消费问题_rabbitmq重复消费使用redis setnx

@Configuration
public class RevisitConfig {

    /\*\*
 \* 创建 direct 队列
 \* \*/
    @Bean
    Queue DirectQueue01() {
        return new Queue("DirectQueue-01",true);
    }

    /\*\*
 \* 创建 direct 交换机
 \* \*/
    @Bean
    DirectExchange DirectExchange01() {
        return new DirectExchange("DirectExchange-01");
    }

    /\*\*
 \* 绑定 direct 队列和交换机
 \* \*/
    @Bean
    Binding bindingDirect01() {
        return BindingBuilder.bind(DirectQueue01()).to(DirectExchange01()).with("DirectRouting01");
    }
}

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

1.4.3 自定义消息应答回调方法

@Component
@Slf4j
public class MyCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    //依赖注入 rabbitTemplate 之后再设置它的回调对象
    // 此注解会在其他注解执行完成后再执行，所以rabbitTemplate先注入，再执行此初始化方法
    @PostConstruct
    public void init() {
        // 设置rabbitTemplate的ConfirmCallBack为我们重写后的类
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }
    /\*\*
 \* 交换机不管是否收到消息都会执行的一个回调方法
 \*
 \* @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
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

1.4.4 数据库对象相关配置：

数据库脚本：

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
6

对象：

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

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

    @TableField("message\_content")
    private String messageContent;
}

1
2
3
4
5
6
7
8
9
10
11
12

mapper：

@Mapper
public interface MessageIdempotentMapper extends BaseMapper<MessageIdempotent> {
}

1
2
3
4

1.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
6
7
8
9
10
11
12
13
14
15
16

访问接口：

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

1
2
3

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

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])

1
2
3

客户端中：

1.4.6 消费者编写：

@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
16
17
18
19
20
21
22

结果：

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 已经消费过！

1
2
3

数据库中：

队列中：

1.5 note Redis 原子性

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

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

2. 消息丢失

2.1 消息丢失的场景

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

2.2 RabbitMQ消息丢失解决方案

2.2.1 针对生产者

1. 方案1 ：开启RabbitMQ事务

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

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

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

}
// 提交事务
channel.txCommit

1
2
3
4
5
6
7
8
9
10
11
12
13

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

2. 方案2： 使用confirm机制

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

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

即第一节MyCallback中：

/\*\*
 \* 交换机不管是否收到消息都会执行的一个回调方法
 \*
 \* @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);
    }
}

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

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方案

为什么设置镜像模式集群，因为队列的内容仅仅存在某一个节点上面，不会存在所有节点上面，所有节点仅仅存放消息结构和元数据。下面自己画了一张图介绍普通集群丢失消息情况：

如果想解决上面途中问题，保证消息不丢失，需要采用HA 镜像模式队列。

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

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

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

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

rabbitmqctl set_policy ha-all "^rock.wechat" '{"ha-mode":"all","ha-sync-mode":"automatic"}'
rabbitmqctl set_policy -p rock ha-all "^rock.wechat" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

1
2
3

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

rabbitmqctl set_policy -p rock ha-exacly "^rock.wechat" \
'{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

1
2
3

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

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

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

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

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）

…(img-AkAjc2Q0-1712858037498)]
[外链图片转存中…(img-y30XRftU-1712858037498)]
[外链图片转存中…(img-xzrOfJle-1712858037498)]
[外链图片转存中…(img-3n9xw3Oi-1712858037498)]

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

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

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
[外链图片转存中…(img-fL572eDf-1712858037499)]