RabbitMQ 避免消息重復消費的方法

更新時間：2024年03月29日 10:42:55 作者：久違の歡喜

消費者端實現冪等性,意味著消息永遠不會消費多次,即使收到了多條一樣的消息,這篇文章給大家分享RabbitMQ 避免消息重復消費的方法,感興趣的朋友一起看看吧

RabbitMQ 如何避免消息重復消費？

冪等性

當消費者消費完消息之后，通常會發(fā)送一個ack應答確認信息給生產者，但是這中間有可能因為網絡中斷等原因，導致生產者未能收到確認消息，由此這條消息將會被重復發(fā)送給其他消費者進行消費，實際上這條消息已經被消費過了，這就是重復消費的問題。

消息冪等性，其實就是保證同一個消息不被消費者重復消費兩次。

如何避免消息重復消費？

消費者端實現冪等性，意味著消息永遠不會消費多次，即使收到了多條一樣的消息。通常有兩種方式來避免消費重復消費：

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

yml 配置文件（公共部分）

spring:
  application:
    name: rabbitmq
  rabbitmq:
    host: 10.0.0.4
    port: 5672
    username: admin
    password: admin
    virtual-host: /
    publisher-confirm-type: correlated # 開啟消息發(fā)布確認機制
    publisher-returns: true # 發(fā)布消息返回監(jiān)聽回調
    # 指定消息確認模式
    listener:
      simple:
        acknowledge-mode: manual
    # 未正確路由的消息發(fā)送到備份隊列
    # 使用備份交換機模式，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

創(chuàng)建本地消息表

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;

創(chuàng)建實體類

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;
}

創(chuàng)建 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> {
}

創(chuàng)建 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 {
    /**
     * 創(chuàng)建 direct 隊列
     * */
    @Bean
    Queue DirectQueue01() {
        return new Queue("DirectQueue-01",true);
    }
    /**
     * 創(chuàng)建 direct 交換機
     * */
    @Bean
    DirectExchange DirectExchange01() {
        return new DirectExchange("DirectExchange-01");
    }
    /**
     * 綁定 direct 隊列和交換機
     * */
    @Bean
    Binding bindingDirect01() {
        return BindingBuilder.bind(DirectQueue01()).to(DirectExchange01()).with("DirectRouting01");
    }
}

自定義消息發(fā)送確認的回調（公共部分）

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 自定義消息發(fā)送確認的回調
 * @Author 聽秋
 * @Date 2022/6/21 19:12
 * @Version 1.0
 **/
@Component
public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * PostConstruct: 用于在依賴關系注入完成之后需要執(zhí)行的方法上，以執(zhí)行任何初始化
     */
    @PostConstruct
    public void init() {
        // 指定 ConfirmCallback
        rabbitTemplate.setConfirmCallback(this);
        // 指定 ReturnCallback
        rabbitTemplate.setReturnsCallback(this);
    }
    /**
     * Confirmation callback.
     *
     * 確認消息是否成功到達交換機中，不管是否到達交換機，該回調都會執(zhí)行
     * 生產者 → 交換機
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("發(fā)布消息的UUID為：" + correlationData.getId());
        if (ack) {
            System.out.println("消息發(fā)布成功！");
        } else {
            System.out.println("消息發(fā)布失??！失敗原因是：" + cause);
        }
    }
    /**
     * Returned message callback.
     *
     * 確認消息是否從交換機成功到達隊列中，失敗將會執(zhí)行，成功則不執(zhí)行
     * 交換機 → 隊列
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.out.println("ReturnsCallback 回調內容：" + returned);
    }
}

創(chuàng)建生產者（公共部分）

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());
        // 消息發(fā)送確認回調
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("DirectExchange-01", routingKey, message, correlationData);
    }
}

發(fā)送消息

控制臺

RabbitMQ 隊列

創(chuàng)建消費者

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 + " 已經消費過！");
        }
    }

消費消息

控制臺

RabbitMQ 隊列

數據庫

基于 Redis 實現消息冪等性

發(fā)送消息

控制臺

RabbitMQ 隊列

創(chuàng)建消費者

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);
        // 消息不存在則創(chuàng)建，返回 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);
        }
    }
}

消費消息

控制臺

RabbitMQ 隊列

Redis

到此這篇關于RabbitMQ 如何避免消息重復消費的文章就介紹到這了,更多相關RabbitMQ 消息重復消費內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

Java結構型設計模式之裝飾模式詳解
裝飾模式（Decorator Pattern）允許向一個現有的對象添加新的功能，同時又不改變其結構。這種類型的設計模式屬于結構型模式，它是作為現有類的一個包裝。這種模式創(chuàng)建了一個裝飾類，用來包裝原有的類，并在保持類方法簽名完整性的前提下，提供了額外的功能
2023-03-03
java算法題解Leetcode763劃分字母區(qū)間示例
這篇文章主要為大家介紹了java算法題解Leetcode763劃分字母區(qū)間示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2023-01-01
使用JPA傳遞參數的方法
這篇文章主要介紹了使用JPA傳遞參數的方法，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2022-06-06
Java多線程之鎖學習(增強版)
這篇文章主要為大家詳細介紹了Java多線程中鎖的相關知識，文中的示例代碼講解詳細，對我們了解線程有一定幫助，需要的可以參考一下
2023-02-02
SpringBoot中的MongoTemplate的各種條件查詢示例詳解
這篇文章主要介紹了SpringBoot中的MongoTemplate的各種條件查詢示例詳解,本文通過示例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借價值,需要的朋友參考下吧
2024-01-01
幾道java循環(huán)練習題（適合新人）
這篇文章主要給大家介紹了幾道java循環(huán)練習題，非常適合剛入門的java新人，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2021-04-04
Spring Cloud 如何保證微服務內安全
這篇文章主要介紹了Spring Cloud 如何保證微服務內安全的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2021-07-07
SpringBoot+jsp項目啟動出現404的解決方法
這篇文章主要介紹了SpringBoot+jsp項目啟動出現404的解決方法，小編覺得挺不錯的，現在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
2019-03-03
Java 讀取文件方法大全
這篇文章主要介紹了Java 讀取文件方法大全,需要的朋友可以參考下
2014-11-11
基于OpenCV與JVM實現矩陣處理圖像
本文主要介紹了Java圖像處理實戰(zhàn)之基于OpenCV與JVM實現矩陣處理圖像。文中的示例代碼講解詳細，對我們學習圖像處理有一定的幫助，感興趣的可以試一試
2022-01-01

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

RabbitMQ 避免消息重復消費的方法

目錄

RabbitMQ 如何避免消息重復消費？

冪等性

如何避免消息重復消費？

基于本地消息表實現消息冪等性

導入 pom.xml 依賴（公共部分）

yml 配置文件（公共部分）

創(chuàng)建本地消息表

創(chuàng)建實體類

創(chuàng)建 mapper 接口

創(chuàng)建 RabbitMQ 配置類（公共部分）

自定義消息發(fā)送確認的回調（公共部分）

創(chuàng)建生產者（公共部分）

發(fā)送消息

創(chuàng)建消費者

消費消息

RabbitMQ 隊列

基于 Redis 實現消息冪等性

發(fā)送消息

創(chuàng)建消費者

消費消息

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

RabbitMQ 避免消息重復消費的方法

目錄

RabbitMQ 如何避免消息重復消費？

冪等性

如何避免消息重復消費？

基于本地消息表實現消息冪等性

導入 pom.xml 依賴（公共部分）

yml 配置文件（公共部分）

創(chuàng)建本地消息表

創(chuàng)建實體類

創(chuàng)建 mapper 接口

創(chuàng)建 RabbitMQ 配置類（公共部分）

自定義消息發(fā)送確認的回調（公共部分）

創(chuàng)建生產者（公共部分）

發(fā)送消息

創(chuàng)建消費者

消費消息

RabbitMQ 隊列

基于 Redis 實現消息冪等性

發(fā)送消息

創(chuàng)建消費者

消費消息

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

RabbitMQ 如何避免消息重復消費？

如何避免消息重復消費？