devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ 如何避免消息重复消费？_rabbitmq怎么保证消息不重复消费

作者：笔触狂放9 | 2024-08-14 03:22:11

赞

踩

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 for Spring Boot</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>

  • 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
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86

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

  • 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

创建本地消息表

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

创建实体类

package com.mq.rabbitmq.vo;


import com.baomidou.mybatisplus.extension.activerecord.Model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @ClassName MessageIdempotent
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/23 16:25
 * @Version 1.0
 **/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageIdempotent extends Model<MessageIdempotent> {

    private String messageId;

    private String messageContent;

}

  • 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

创建 mapper 接口

package com.mq.rabbitmq.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.mq.rabbitmq.vo.MessageIdempotent;
import org.apache.ibatis.annotations.Mapper;

/**
 * @ClassName MessageIdempotentMapper
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/23 19:03
 * @Version 1.0
 **/
@Mapper
public interface MessageIdempotentMapper extends BaseMapper<MessageIdempotent> {
}

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

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

package com.mq.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @ClassName DirectConfig
 * @Description TODO direct 交换机配置类
 * @Author 听秋
 * @Date 2022/6/17 16:30
 * @Version 1.0
 **/
@Configuration
public class ExchangeConfig {

    /**
     * 创建 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
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

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

package com.mq.rabbitmq.callback;

import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @ClassName MyConfirmCallback
 * @Description TODO 自定义消息发送确认的回调
 * @Author 听秋
 * @Date 2022/6/21 19:12
 * @Version 1.0
 **/
@Component
public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化
     */
    @PostConstruct
    public void init() {
        // 指定 ConfirmCallback
        rabbitTemplate.setConfirmCallback(this);
        // 指定 ReturnCallback
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * Confirmation callback.
     *
     * 确认消息是否成功到达交换机中，不管是否到达交换机，该回调都会执行
     * 生产者 → 交换机
     */
    @Override
    public void confirm(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.
     *
     * 确认消息是否从交换机成功到达队列中，失败将会执行，成功则不执行
     * 交换机 → 队列
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.out.println("ReturnsCallback 回调内容：" + returned);
    }
}

  • 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
  • 61

创建生产者（公共部分）

package com.mq.rabbitmq.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;


/**
 * @ClassName DirectController
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/17 16:48
 * @Version 1.0
 **/
@RestController
@Slf4j
public class QueueController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 消息幂等性
     * */
    @GetMapping("/sendMessage")
    public void sendMessage(String msg, String routingKey, 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", routingKey, message, correlationData);
    }
}

  • 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

发送消息

控制台

在这里插入图片描述
RabbitMQ 队列

在这里插入图片描述

创建消费者

package com.mq.rabbitmq.controller;

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.mq.rabbitmq.vo.MessageIdempotent;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @ClassName DirectReceiver
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/21 10:40
 * @Version 1.0
 **/
@Component
@Slf4j
public class Consumer {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

	/**
     * 基于本地消息表实现消息幂等性
     * @param message
     */
    @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();
        QueryWrapper<MessageIdempotent> queryWrapper = new QueryWrapper<>();
        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 + " 已经消费过！");
        }
    }

  • 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

消费消息

控制台

在这里插入图片描述

RabbitMQ 队列

在这里插入图片描述
数据库

在这里插入图片描述

基于 Redis 实现消息幂等性

发送消息

控制台

在这里插入图片描述

RabbitMQ 队列

在这里插入图片描述

创建消费者

package com.mq.rabbitmq.controller;

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.mq.rabbitmq.vo.MessageIdempotent;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @ClassName DirectReceiver
 * @Description TODO
 * @Author 听秋
 * @Date 2022/6/21 10:40
 * @Version 1.0
 **/
@Component
@Slf4j
public class Consumer {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

	/**
     * 基于 redis 实现消息幂等性
     * @param message
     */
    //@RabbitHandler
    //@RabbitListener(queues = "DirectQueue-01")
    public void receiveMessage01(Message message, Channel channel) throws IOException {
        String messageId = message.getMessageProperties().getMessageId();
        String messageContent = new String(message.getBody(), StandardCharsets.UTF_8);

        // 消息不存在则创建，返回 true
        Boolean 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);
        }
    }
}

  • 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

消费消息

控制台

在这里插入图片描述

RabbitMQ 队列

在这里插入图片描述

Redis

在这里插入图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/977782
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号