RabbitMQ 避免消息重復(fù)消費的方法

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

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

RabbitMQ 如何避免消息重復(fù)消費？

冪等性

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

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

如何避免消息重復(fù)消費？

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

方式1: 消息全局 ID 或者寫個唯一標(biāo)識(如時間戳、UUID 等) ：每次消費消息之前根據(jù)消息 id 去判斷該消息是否已消費過，如果已經(jīng)消費過，則不處理這條消息，否則正常消費消息，并且進行入庫操作。(消息全局 ID 作為數(shù)據(jù)庫表的主鍵，防止重復(fù))
方式2: 利用 Redis 的 setnx 命令：給消息分配一個全局 ID，消費該消息時，先去 Redis 中查詢有沒消費記錄，無則以鍵值對形式寫入 Redis ，有則不消費該消息。

基于本地消息表實現(xiàn)消息冪等性

導(dǎo)入 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ā)布確認(rèn)機制
    publisher-returns: true # 發(fā)布消息返回監(jiān)聽回調(diào)
    # 指定消息確認(rèn)模式
    listener:
      simple:
        acknowledge-mode: manual
    # 未正確路由的消息發(fā)送到備份隊列
    # 使用備份交換機模式，mandatory 將無效，即就算 mandatory設(shè) 置為 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 插件打印日志，打印查詢結(jié)果
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 '消息內(nèi)容',
  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ā)送確認(rèn)的回調(diào)（公共部分）

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ā)送確認(rèn)的回調(diào)
 * @Author 聽秋
 * @Date 2022/6/21 19:12
 * @Version 1.0
 **/
@Component
public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * PostConstruct: 用于在依賴關(guān)系注入完成之后需要執(zhí)行的方法上，以執(zhí)行任何初始化
     */
    @PostConstruct
    public void init() {
        // 指定 ConfirmCallback
        rabbitTemplate.setConfirmCallback(this);
        // 指定 ReturnCallback
        rabbitTemplate.setReturnsCallback(this);
    }
    /**
     * Confirmation callback.
     *
     * 確認(rèn)消息是否成功到達交換機中，不管是否到達交換機，該回調(diào)都會執(zhí)行
     * 生產(chǎn)者 → 交換機
     */
    @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.
     *
     * 確認(rèn)消息是否從交換機成功到達隊列中，失敗將會執(zhí)行，成功則不執(zhí)行
     * 交換機 → 隊列
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.out.println("ReturnsCallback 回調(diào)內(nèi)容：" + returned);
    }
}

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

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("生產(chǎn)消息:" + message.toString());
        // 消息發(fā)送確認(rèn)回調(diào)
        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;
	/**
     * 基于本地消息表實現(xiàn)消息冪等性
     * @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 + " 消息內(nèi)容：" + messageContent);
            // 消息確認(rèn)
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            log.info("消息 " + messageId + " 已經(jīng)消費過！");
        }
    }

消費消息

控制臺

RabbitMQ 隊列

數(shù)據(jù)庫

基于 Redis 實現(xiàn)消息冪等性

發(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 實現(xiàn)消息冪等性
     * @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 + " 已經(jīng)消費過");
        } else {
        	// 消息確認(rèn)
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("DirectQueue-01-消費者收到消息，消息ID：" + messageId + " 消息內(nèi)容：" + messageContent);
        }
    }
}

消費消息

控制臺

RabbitMQ 隊列

Redis

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

您可能感興趣的文章:

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

RabbitMQ 避免消息重復(fù)消費的方法

目錄

RabbitMQ 如何避免消息重復(fù)消費？

冪等性

如何避免消息重復(fù)消費？

基于本地消息表實現(xiàn)消息冪等性

導(dǎo)入 pom.xml 依賴（公共部分）

yml 配置文件（公共部分）

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

創(chuàng)建實體類

創(chuàng)建 mapper 接口

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

自定義消息發(fā)送確認(rèn)的回調(diào)（公共部分）

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

發(fā)送消息

創(chuàng)建消費者

消費消息

RabbitMQ 隊列

基于 Redis 實現(xiàn)消息冪等性

發(fā)送消息

創(chuàng)建消費者

消費消息

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

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

RabbitMQ 避免消息重復(fù)消費的方法

目錄

RabbitMQ 如何避免消息重復(fù)消費？

冪等性

如何避免消息重復(fù)消費？

基于本地消息表實現(xiàn)消息冪等性

導(dǎo)入 pom.xml 依賴（公共部分）

yml 配置文件（公共部分）

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

創(chuàng)建實體類

創(chuàng)建 mapper 接口

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

自定義消息發(fā)送確認(rèn)的回調(diào)（公共部分）

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

發(fā)送消息

創(chuàng)建消費者

消費消息

RabbitMQ 隊列

基于 Redis 實現(xiàn)消息冪等性

發(fā)送消息

創(chuàng)建消費者

消費消息

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

如何避免消息重復(fù)消費？