• RabbitMQ 的死信交换机和备份交换机

    文章目录

    RabbitMQ 的死信交换机和备份交换机

    死信交换机（Dead-Letter-Exchange）

    当消息在一个队列中由于过期、被拒绝等原因变成死信 (dead message) 之后，它能被重新被发送到另一个交换器中，这个交换器就是死信交换机，绑定死信交换机的队列就称之为死信队列。

    判断一个消息是否是死信消息（Dead Message）的依据：

    • 消息被拒绝（Basic.Reject 或 Basic.Nack）并且设置 requeue 参数的值为 false。
    • 消息过期。消息过期时间设置主要有两种方式：

    1.设置队列的过期时间，队列中所有的消息都存在相同的过期时间（在队列申明的时候使用 x-message-ttl 参数，单位为 毫秒）。

    2.单独设置某个消息的过期时间，每条消息的过期时间都不一样；（设置消息属性的 expiration 参数的值，单位为 毫秒）。

    3.如果同时使用了两种方式设置过期时间，以两者之间较小的那个数值为准。

    • 队列已满（队列满了，无法再添加消息到 mq 中）。

    使用方法：申明队列的时候设置 x-dead-letter-exchange 参数。

    备份交换机(alternate-exchange)

    未被正确路由的消息将会经过此交换器。

    使用方法：申明交换器的时候设置 alternate-exchange 参数。

    死信交换机和备份交换机的应用

    在这里插入图片描述

    导入 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

    创建 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 {

    /*------------------------备份队列和备份交换机（建议用 fanout 交换机）---------------------------*/

    /**
     * 创建 direct 队列
     * */
    @Bean("backupQueue")
    Queue FanoutQueue01() {
        return new Queue("FanoutQueue-01");
    }

    /**
     * 创建 direct 交换机
     * */
    @Bean("backupExchange")
    FanoutExchange FanoutExchange01() {
        return new FanoutExchange("FanoutExchange-01");
    }

    /**
     * 绑定 direct 队列和交换机
     * */
    @Bean
    Binding bindingFanout01(@Qualifier("backupQueue") Queue queue, @Qualifier("backupExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    /*------------------------死信队列和死信交换机（建议用 fanout 交换机）---------------------------*/

    /**
     * 创建 direct 队列
     * */
    @Bean
    Queue FanoutQueue02() {
        return new Queue("FanoutQueue-02",true);
    }

    /**
     * 创建 direct 交换机
     * */
    @Bean
    FanoutExchange FanoutExchange02() {
        return new FanoutExchange("FanoutExchange-02");
    }

    /**
     * 绑定 direct 队列和交换机
     * */
    @Bean
    Binding bindingDirect05() {
        return BindingBuilder.bind(FanoutQueue02()).to(FanoutExchange02());
    }

    /*------------------------正常交换机和队列---------------------------*/

    /**
     * 创建 direct 队列
     * */
    @Bean
    Queue Queue() {
        Map<String, Object> arguments = new HashMap<>(10);
        arguments.put("x-dead-letter-exchange", "FanoutExchange-02");
        return new Queue("Queue",true, false, false, arguments);
    }

    /**
     * 创建 direct 交换机
     * */
    @Bean
    Exchange Exchange() {
        Map<String, Object> arguments = new HashMap<>(10);
        arguments.put("alternate-exchange", "FanoutExchange-01");
        return new DirectExchange("Exchange", true, false, arguments);
    }

    /**
     * 绑定 direct 队列和交换机
     * */
    @Bean
    Binding binding() {
        return BindingBuilder.bind(Queue()).to(Exchange()).with("routingKey").noargs();
    }

    • 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
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101

    自定义消息发送确认的回调

    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("/sendMsg")
    public void sendMsg(String msg, String routingKey) {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text/plain");
        messageProperties.setContentEncoding("utf-8");
        Message message = new Message(msg.getBytes(), messageProperties);
        System.out.println("生产消息:" + message.toString());

        // 消息发送确认回调
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        // convertAndSend 方法默认持久化(底层 DeliveryMode = 2)
        rabbitTemplate.convertAndSend("Exchange", 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
    • 47
    • 48

    创建消费者

    创建消费者

    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 {

	/**
     * 死信队列
     * @param msg
     * @param channel
     * @param message
     * @throws IOException
     */
    @RabbitListener(queues = "Queue")
    public void receiveMessage04(String msg, Channel channel, Message message) throws IOException {
        if (msg.equals("死信交换机验证")) {
            log.info("死信消息：" + msg);
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            log.info("【Consumer】成功接收到消息：" + msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    • 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

    Direct 交换机验证

    通过 postman 调用接口 http://localhost:8080/sendMsg?msg=Direct 交换机&routingKey=routingKey

    在这里插入图片描述

    备份交换机验证

    通过 postman 调用接口 http://localhost:8080/sendMsg?msg=备份交换机&routingKey=routing

    在这里插入图片描述

    死信交换机验证

    通过 postman 调用接口 http://localhost:8080/sendMsg?msg=死信交换机&routingKey=routingKey

    在这里插入图片描述

  • 相关阅读:
    ThingsBoard的数据分析-自定义节点来订阅kafka stream的消息
    关于 SAP UI5 SimpleForm 控件里的 ColumnsL 和 labelSpanXL 的测试
    CodeForces_1732A
    在.NET 8 RC1 版本中 MAUI、ASP.NET Core 和 EF8 的新特性
    open3d 0.17.0的open3d.visualization.ViewControl类有bug
    java项目-第149期ssm师生交流平台_java毕业设计_计算机毕业设计
    string的学习
    C++继承同名静态成员处理方式
    linux 实时调度实现
    职场数据汇总，数据汇报，动态大屏可视化制作与数据分析
  • 原文地址：https://blog.csdn.net/TQ20160412/article/details/125434119