RabbitMQ 如何避免消息重复消费？

文章目录

RabbitMQ 如何避免消息重复消费？

幂等性

当消费者消费完消息之后，通常会发送一个ack应答确认信息给生产者，但是这中间有可能因为网络中断等原因，导致生产者未能收到确认消息，由此这条消息将会被 重复发送给其他消费者进行消费，实际上这条消息已经被消费过了，这就是重复消费的问题。

消息幂等性，其实就是保证同一个消息不被消费者重复消费两次。

如何避免消息重复消费？

消费者端实现幂等性，意味着消息永远不会消费多次，即使收到了多条一样的消息。通常有两种方式来避免消费重复消费：

• 方式1: 消息全局 ID 或者写个唯一标识(如时间戳、UUID 等) ：每次消费消息之前根据消息 id 去判断该消息是否已消费过，如果已经消费过，则不处理这条消息，否则正常消费消息，并且进行入库操作。(消息全局 ID 作为数据库表的主键，防止重复)
• 方式2: 利用 Redis 的 setnx 命令：给消息分配一个全局 ID，消费该消息时，先去 Redis 中查询有没消费记录，无则以键值对形式写入 Redis ，有则不消费该消息。

基于本地消息表实现消息幂等性

导入 pom.xml 依赖（公共部分）

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.0</version><relativePath/><!-- lookup parent from repository --></parent><groupId>com.mq</groupId><artifactId>rabbitmq</artifactId><version>0.0.1-SNAPSHOT</version><name>rabbitmq</name><description>Demo project forSpringBoot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp --><!--springboot rabbit 启动器--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.7.0</version></dependency><!-- https://mvnrepository.com/artifact/org.projectlombok/lombok --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.24</version><scope>provided</scope></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><!-- https://mvnrepository.com/artifact/cn.hutool/hutool-all --><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.3</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.4.0</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus</artifactId><version>3.4.0</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>2.6.7</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

yml 配置文件（公共部分）

spring:
  application:
    name: rabbitmq
  rabbitmq:
    host: 10.0.0.4
    port: 5672
    username: admin
    password: admin
    virtual-host: /
    publisher-confirm-type: correlated # 开启消息发布确认机制
    publisher-returns: true# 发布消息返回监听回调# 指定消息确认模式
    listener:
      simple:
        acknowledge-mode: manual
    # 未正确路由的消息发送到备份队列# 使用备份交换机模式，mandatory 将无效，即就算 mandatory设 置为 false，路由失败的消息同样会被投递到绑定的备份交换机
    template:
      mandatory: true
  redis:
    host: 10.0.0.4
    database: 0
    port: 6379
    timeout: 300ms
  datasource:
    url: jdbc:mysql://10.0.0.4:3306/Test?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai
    username: root
    password: root

server:
  port: 8080# mybatis 插件打印日志，打印查询结果
mybatis-plus:
  configuration:
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

创建本地消息表

CREATETABLE`message_idempotent`(`message_id`varchar(50)NOTNULLCOMMENT'消息ID',`message_content`varchar(2000)DEFAULTNULLCOMMENT'消息内容',PRIMARYKEY(`message_id`))ENGINE=InnoDBDEFAULTCHARSET=utf8;

创建实体类

packagecom.mq.rabbitmq.vo;importcom.baomidou.mybatisplus.extension.activerecord.Model;importlombok.AllArgsConstructor;importlombok.Data;importlombok.NoArgsConstructor;/**
 * @ClassName MessageIdempotent
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/23 16:25
 * @Version 1.0
 **/@Data@NoArgsConstructor@AllArgsConstructorpublicclassMessageIdempotentextendsModel<MessageIdempotent>{privateString messageId;privateString messageContent;}

创建 mapper 接口

packagecom.mq.rabbitmq.mapper;importcom.baomidou.mybatisplus.core.mapper.BaseMapper;importcom.mq.rabbitmq.vo.MessageIdempotent;importorg.apache.ibatis.annotations.Mapper;/**
 * @ClassName MessageIdempotentMapper
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/23 19:03
 * @Version 1.0
 **/@MapperpublicinterfaceMessageIdempotentMapperextendsBaseMapper<MessageIdempotent>{}

创建 RabbitMQ 配置类（公共部分）

packagecom.mq.rabbitmq.config;importorg.springframework.amqp.core.*;importorg.springframework.beans.factory.annotation.Qualifier;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importjava.util.HashMap;importjava.util.Map;/**
 * @ClassName DirectConfig
 * @Description TODO direct 交换机配置类
 * @Author 听秋
 * @Date 2022/6/17 16:30
 * @Version 1.0
 **/@ConfigurationpublicclassExchangeConfig{/**
     * 创建 direct 队列
     * */@BeanQueueDirectQueue01(){returnnewQueue("DirectQueue-01",true);}/**
     * 创建 direct 交换机
     * */@BeanDirectExchangeDirectExchange01(){returnnewDirectExchange("DirectExchange-01");}/**
     * 绑定 direct 队列和交换机
     * */@BeanBindingbindingDirect01(){returnBindingBuilder.bind(DirectQueue01()).to(DirectExchange01()).with("DirectRouting01");}}

自定义消息发送确认的回调（公共部分）

packagecom.mq.rabbitmq.callback;importorg.springframework.amqp.core.ReturnedMessage;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;importjavax.annotation.PostConstruct;/**
 * @ClassName MyConfirmCallback
 * @Description TODO 自定义消息发送确认的回调
 * @Author 听秋
 * @Date 2022/6/21 19:12
 * @Version 1.0
 **/@ComponentpublicclassMyConfirmCallbackimplementsRabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{@AutowiredprivateRabbitTemplate rabbitTemplate;/**
     * PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化
     */@PostConstructpublicvoidinit(){// 指定 ConfirmCallback
        rabbitTemplate.setConfirmCallback(this);// 指定 ReturnCallback
        rabbitTemplate.setReturnsCallback(this);}/**
     * Confirmation callback.
     *
     * 确认消息是否成功到达交换机中，不管是否到达交换机，该回调都会执行
     * 生产者 → 交换机
     */@Overridepublicvoidconfirm(CorrelationData correlationData,boolean ack,String cause){System.out.println("发布消息的UUID为："+ correlationData.getId());if(ack){System.out.println("消息发布成功！");}else{System.out.println("消息发布失败！失败原因是："+ cause);}}/**
     * Returned message callback.
     *
     * 确认消息是否从交换机成功到达队列中，失败将会执行，成功则不执行
     * 交换机 → 队列
     */@OverridepublicvoidreturnedMessage(ReturnedMessage returned){System.out.println("ReturnsCallback 回调内容："+ returned);}}

创建生产者（公共部分）

packagecom.mq.rabbitmq.controller;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.core.MessageDeliveryMode;importorg.springframework.amqp.core.MessagePostProcessor;importorg.springframework.amqp.core.MessageProperties;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;importjava.util.UUID;/**
 * @ClassName DirectController
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/17 16:48
 * @Version 1.0
 **/@RestController@Slf4jpublicclassQueueController{@AutowiredprivateRabbitTemplate rabbitTemplate;/**
     * 消息幂等性
     * */@GetMapping("/sendMessage")publicvoidsendMessage(String msg,String routingKey,String id){MessageProperties messageProperties =newMessageProperties();
        messageProperties.setMessageId(id);
        messageProperties.setContentType("text/plain");
        messageProperties.setContentEncoding("utf-8");Message message =newMessage(msg.getBytes(), messageProperties);
        log.info("生产消息:"+ message.toString());// 消息发送确认回调CorrelationData correlationData =newCorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("DirectExchange-01", routingKey, message, correlationData);}}

发送消息

控制台

RabbitMQ 队列

创建消费者

packagecom.mq.rabbitmq.controller;importcn.hutool.core.util.ObjectUtil;importcom.baomidou.mybatisplus.core.conditions.query.QueryWrapper;importcom.mq.rabbitmq.vo.MessageIdempotent;importcom.rabbitmq.client.Channel;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.stereotype.Component;importjava.io.IOException;importjava.nio.charset.StandardCharsets;/**
 * @ClassName DirectReceiver
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/21 10:40
 * @Version 1.0
 **/@Component@Slf4jpublicclassConsumer{@AutowiredprivateStringRedisTemplate stringRedisTemplate;/**
     * 基于本地消息表实现消息幂等性
     * @param message
     */@RabbitListener(queues ="DirectQueue-01")publicvoidreceiveMessage02(Message message,Channel channel)throwsIOException{String messageId = message.getMessageProperties().getMessageId();String messageContent =newString(message.getBody(),StandardCharsets.UTF_8);MessageIdempotent messageIdempotent =newMessageIdempotent();QueryWrapper<MessageIdempotent> queryWrapper =newQueryWrapper<>();
        queryWrapper.eq("message_id", messageId);MessageIdempotent msg = messageIdempotent.selectOne(queryWrapper);if(ObjectUtil.isNull(msg)){
            messageIdempotent.setMessageId(messageId);
            messageIdempotent.setMessageContent(messageContent);
            messageIdempotent.insert();
            log.info("DirectQueue-01-消费者收到消息，消息ID："+ messageId +" 消息内容："+ messageContent);// 消息确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}else{
            log.info("消息 "+ messageId +" 已经消费过！");}}

消费消息

控制台

RabbitMQ 队列

数据库

基于 Redis 实现消息幂等性

发送消息

控制台

RabbitMQ 队列

创建消费者

packagecom.mq.rabbitmq.controller;importcn.hutool.core.util.ObjectUtil;importcom.baomidou.mybatisplus.core.conditions.query.QueryWrapper;importcom.mq.rabbitmq.vo.MessageIdempotent;importcom.rabbitmq.client.Channel;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.stereotype.Component;importjava.io.IOException;importjava.nio.charset.StandardCharsets;/**
 * @ClassName DirectReceiver
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/21 10:40
 * @Version 1.0
 **/@Component@Slf4jpublicclassConsumer{@AutowiredprivateStringRedisTemplate stringRedisTemplate;/**
     * 基于 redis 实现消息幂等性
     * @param message
     *///@RabbitHandler//@RabbitListener(queues = "DirectQueue-01")publicvoidreceiveMessage01(Message message,Channel channel)throwsIOException{String messageId = message.getMessageProperties().getMessageId();String messageContent =newString(message.getBody(),StandardCharsets.UTF_8);// 消息不存在则创建，返回 trueBoolean exist = stringRedisTemplate.opsForValue().setIfAbsent(messageId, messageContent);if(!exist){
            log.info("消息 "+ messageId +" 已经消费过");}else{// 消息确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            log.info("DirectQueue-01-消费者收到消息，消息ID："+ messageId +" 消息内容："+ messageContent);}}}

消费消息

控制台

RabbitMQ 队列

Redis