欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

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

 更新時間：2024年03月29日 10:42:55   作者：久違の歡喜  
消費(fèi)者端實(shí)現(xiàn)冪等性,意味著消息永遠(yuǎn)不會消費(fèi)多次,即使收到了多條一樣的消息,這篇文章給大家分享RabbitMQ 避免消息重復(fù)消費(fèi)的方法,感興趣的朋友一起看看吧

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

冪等性

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

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

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

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

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

基于本地消息表實(shí)現(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)機(jī)制
    publisher-returns: true # 發(fā)布消息返回監(jiān)聽回調(diào)
    # 指定消息確認(rèn)模式
    listener:
      simple:
        acknowledge-mode: manual
    # 未正確路由的消息發(fā)送到備份隊(duì)列
    # 使用備份交換機(jī)模式，mandatory 將無效，即就算 mandatory設(shè) 置為 false，路由失敗的消息同樣會被投遞到綁定的備份交換機(jī)
    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)建實(shí)體類

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 交換機(jī)配置類
 * @Author 聽秋
 * @Date 2022/6/17 16:30
 * @Version 1.0
 **/
@Configuration
public class ExchangeConfig {
    /**
     * 創(chuàng)建 direct 隊(duì)列
     * */
    @Bean
    Queue DirectQueue01() {
        return new Queue("DirectQueue-01",true);
    }
    /**
     * 創(chuàng)建 direct 交換機(jī)
     * */
    @Bean
    DirectExchange DirectExchange01() {
        return new DirectExchange("DirectExchange-01");
    }
    /**
     * 綁定 direct 隊(duì)列和交換機(jī)
     * */
    @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)消息是否成功到達(dá)交換機(jī)中，不管是否到達(dá)交換機(jī)，該回調(diào)都會執(zhí)行
     * 生產(chǎn)者 → 交換機(jī)
     */
    @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)消息是否從交換機(jī)成功到達(dá)隊(duì)列中，失敗將會執(zhí)行，成功則不執(zhí)行
     * 交換機(jī) → 隊(duì)列
     */
    @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 隊(duì)列

創(chuàng)建消費(fèi)者

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;
	/**
     * 基于本地消息表實(shí)現(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-消費(fèi)者收到消息，消息ID：" + messageId + " 消息內(nèi)容：" + messageContent);
            // 消息確認(rèn)
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            log.info("消息 " + messageId + " 已經(jīng)消費(fèi)過！");
        }
    }

消費(fèi)消息

控制臺

RabbitMQ 隊(duì)列


數(shù)據(jù)庫

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

發(fā)送消息

控制臺

RabbitMQ 隊(duì)列

創(chuàng)建消費(fèi)者

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 實(shí)現(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)消費(fèi)過");
        } else {
        	// 消息確認(rèn)
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("DirectQueue-01-消費(fèi)者收到消息，消息ID：" + messageId + " 消息內(nèi)容：" + messageContent);
        }
    }
}

消費(fèi)消息

控制臺

RabbitMQ 隊(duì)列

Redis

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

相關(guān)文章

最新評論