RabbitMQ之发布确认

王远洋

 关注

阅读 161

2022-12-11

目录

8.1. 发布确认 springboot 版本

8.1.1. 确认机制方案

8.1.2. 代码架构图

8.1.3. 配置文件

8.1.4. 添加配置类

8.1.5. 消息生产者

8.1.6. 回调接口

8.1.7. 消息消费者

8.1.8. 结果分析

8.2. 回退消息

8.2.1. Mandatory 参数

8.2.2. 消息生产者代码

8.2.3. 回调接口

8.2.4. 结果分析

8.3. 备份交换机

8.3.1. 代码架构图

8.3.2. 修改配置类

8.3.3. 报警消费者

8.3.4. 测试注意事项

8.3.5. 结果分析

在生产环境中由于一些不明原因，导致 rabbitmq 重启，在 RabbitMQ 重启期间生产者消息投递失败，

导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？

特别是在这样比较极端的情况，RabbitMQ 集群不可用的时候，无法投递的消息该如何处理呢

8.1. 发布确认 springboot 版本

8.1.1. 确认机制方案

8.1.2. 代码架构图

8.1.3. 配置文件

在配置文件当中需要添加

⚫ NONE

禁用发布确认模式，是默认值

⚫ CORRELATED

发布消息成功到交换器后会触发回调方法

⚫ SIMPLE

经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，

其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法

等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是

waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker

8.1.4. 添加配置类

package com.xingchen.mq.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;

/**
 * @author xingchen
 * @version V1.0
 *  * 发布确认兜底方案  添加缓存测试
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Configuration
public class ConfirmConfig {

    //交换机
    private static final String CONFIRM_EXCHANGE_NAME = "confirm-exchange";
    //队列

    private static final String CONFIRM_QUEUE_NAME = "confirm-queue";
    //ROUTING_KEY

    private static final String CONFIRM_ROUTING_KEY = "confirm-key";

    /**
     * 备份交换机
     */
    private static final String BACKUP_EXCHANGE_NAME = "backup-exchange";
    /**
     * 备份队列
     */
    private static final String BACKUP_QUEUE_NAME = "backup-queue";
    /**
     * 报警队列
     */
    private static final String WARNING_QUEUE_NAME = "warning-queue";

    @Bean("confirmExchange")
    public DirectExchange directExchange() {
        /**确认交换机配置备份交换机 以确保宕机后将消息转发到备份交换机*/
        return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
                .withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME).build();
    }

    @Bean("backupExchange")
    public FanoutExchange backupExchange() {
        return new FanoutExchange(BACKUP_EXCHANGE_NAME);
    }

    @Bean("confirmQueue")
    public Queue confirmQueue() {
        HashMap<String, Object> map = new HashMap<>(8);
        return new Queue(CONFIRM_QUEUE_NAME, false, false, false, map);
    }

    @Bean("backupQueue")
    public Queue backupQueue() {
        HashMap<String, Object> map = new HashMap<>(8);
        return new Queue(BACKUP_QUEUE_NAME, false, false, false, map);
    }

    @Bean("warningQueue")
    public Queue warningQueue() {
        HashMap<String, Object> map = new HashMap<>(8);
        return new Queue(WARNING_QUEUE_NAME, false, false, false, map);
    }

    @Bean
    public Binding queueConfirmBindingExchange(@Qualifier("confirmQueue") Queue queue,
                                               @Qualifier("confirmExchange") Exchange exchange) {

        return BindingBuilder.bind(queue).to(exchange).with(CONFIRM_ROUTING_KEY).noargs();
    }

    @Bean
    public Binding backupConfirmBindingExchange(@Qualifier("backupQueue") Queue queue,
                                                @Qualifier("backupExchange") FanoutExchange exchange) {

        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public Binding warningConfirmBindingExchange(@Qualifier("warningQueue") Queue queue,
                                                 @Qualifier("backupExchange") FanoutExchange exchange) {

        return BindingBuilder.bind(queue).to(exchange);
    }
}

8.1.5. 消息生产者

package com.xingchen.mq.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @author xingchen
 * @version V1.0
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Slf4j
@RestController
@RequestMapping("/confirm")
public class ConfirmController {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendConfirm/{msg}")
    public void sendConfirmMessage(@PathVariable("msg") String msg) {
        /**声明回调的形参*/

        CorrelationData correlationData = new CorrelationData("1");
        rabbitTemplate.convertAndSend("confirm-exchange", "confirm-key", msg, correlationData);
        log.info("发送信息为:" + msg);
    }
}

8.1.6. 回调接口

package com.xingchen.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @author xingchen
 * @version V1.0
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 该注解会在其他注解执行完毕之后，进行一个属性的注入，必须将该类注入到rabbitTemplate的内部类中
     * 内部类就是这个ConfirmCallback
     */
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        /**同时需要注入队列回退接口*/
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * @param correlationData 包含了消息的ID和其他数据信息 这个需要在发送方创建，否则没有
     * @param ack             返回的一个交换机确认状态 true 为确认 false 为未确认
     * @param cause           未确认的一个原因，如果ack为true的话，此值为null
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("消息发送成功，id 是{} ", id);
        } else {
            log.info("消息发送失败，原因 是{} id 为{}", cause, id);
        }
    }
    /**
     * 可以在消息传递过程中，如果交换机遇到不可路由的情况，会将消息返回给生产者
     *
     * @param message    消息
     * @param replyCode  回复状态码
     * @param replyText  退回原因
     * @param exchange   交换机
     * @param routingKey 路由Key
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("消息{}，被交换机{}退回，路由Key是{}，退回原因是{}", new String(message.getBody()), exchange
                , routingKey, replyText);
    }
}

8.1.7. 消息消费者

package com.xingchen.mq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author xingchen
 * @version V1.0
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Component
@Slf4j
public class ConfirmConsumer {
    @RabbitListener(queues = {"confirm-queue"})
    public void receiveMsg(Message message) {
        log.info("接收到的消息为: " + new String(message.getBody()));
    }
}

8.1.8. 结果分析

可以看到，发送了两条消息，第一条消息的 RoutingKey 为 "key1"，第二条消息的 RoutingKey 为

"key2"，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为

第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条

消息被直接丢弃了。

8.2. 回退消息

8.2.1. Mandatory 参数

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如

果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。那么如何

让无法被路由的消息帮我想办法处理一下？最起码通知我一声，我好自己处理啊。通过设置 mandatory 参

数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

8.2.2. 消息生产者代码

package com.xingchen.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @author xingchen
 * @version V1.0
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 该注解会在其他注解执行完毕之后，进行一个属性的注入，必须将该类注入到rabbitTemplate的内部类中
     * 内部类就是这个ConfirmCallback
     */
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        /**同时需要注入队列回退接口*/
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * @param correlationData 包含了消息的ID和其他数据信息 这个需要在发送方创建，否则没有
     * @param ack             返回的一个交换机确认状态 true 为确认 false 为未确认
     * @param cause           未确认的一个原因，如果ack为true的话，此值为null
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("消息发送成功，id 是{} ", id);
        } else {
            log.info("消息发送失败，原因 是{} id 为{}", cause, id);
        }
    }

    /**
     * 可以在消息传递过程中，如果交换机遇到不可路由的情况，会将消息返回给生产者
     *
     * @param message    消息
     * @param replyCode  回复状态码
     * @param replyText  退回原因
     * @param exchange   交换机
     * @param routingKey 路由Key
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("消息{}，被交换机{}退回，路由Key是{}，退回原因是{}", new String(message.getBody()), exchange
                , routingKey, replyText);
    }
}

8.2.3. 回调接口

package com.xingchen.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @author xingchen
 * @version V1.0
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 该注解会在其他注解执行完毕之后，进行一个属性的注入，必须将该类注入到rabbitTemplate的内部类中
     * 内部类就是这个ConfirmCallback
     */
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        /**同时需要注入队列回退接口*/
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * @param correlationData 包含了消息的ID和其他数据信息 这个需要在发送方创建，否则没有
     * @param ack             返回的一个交换机确认状态 true 为确认 false 为未确认
     * @param cause           未确认的一个原因，如果ack为true的话，此值为null
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("消息发送成功，id 是{} ", id);
        } else {
            log.info("消息发送失败，原因 是{} id 为{}", cause, id);
        }
    }

    /**
     * 可以在消息传递过程中，如果交换机遇到不可路由的情况，会将消息返回给生产者
     *
     * @param message    消息
     * @param replyCode  回复状态码
     * @param replyText  退回原因
     * @param exchange   交换机
     * @param routingKey 路由Key
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("消息{}，被交换机{}退回，路由Key是{}，退回原因是{}", new String(message.getBody()), exchange
                , routingKey, replyText);
    }
}

8.2.4. 结果分析

8.3. 备份交换机

8.3.1. 代码架构图

8.3.2. 修改配置类

package com.xingchen.mq.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;

/**
 * @author xingchen
 * @version V1.0
 *  * 发布确认兜底方案  添加缓存测试
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Configuration
public class ConfirmConfig {

    //交换机
    private static final String CONFIRM_EXCHANGE_NAME = "confirm-exchange";
    //队列

    private static final String CONFIRM_QUEUE_NAME = "confirm-queue";
    //ROUTING_KEY

    private static final String CONFIRM_ROUTING_KEY = "confirm-key";

    /**
     * 备份交换机
     */
    private static final String BACKUP_EXCHANGE_NAME = "backup-exchange";
    /**
     * 备份队列
     */
    private static final String BACKUP_QUEUE_NAME = "backup-queue";
    /**
     * 报警队列
     */
    private static final String WARNING_QUEUE_NAME = "warning-queue";

    @Bean("confirmExchange")
    public DirectExchange directExchange() {
        /**确认交换机配置备份交换机 以确保宕机后将消息转发到备份交换机*/
        return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
                .withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME).build();
    }

    @Bean("backupExchange")
    public FanoutExchange backupExchange() {
        return new FanoutExchange(BACKUP_EXCHANGE_NAME);
    }

    @Bean("confirmQueue")
    public Queue confirmQueue() {
        HashMap<String, Object> map = new HashMap<>(8);
        return new Queue(CONFIRM_QUEUE_NAME, false, false, false, map);
    }

    @Bean("backupQueue")
    public Queue backupQueue() {
        HashMap<String, Object> map = new HashMap<>(8);
        return new Queue(BACKUP_QUEUE_NAME, false, false, false, map);
    }

    @Bean("warningQueue")
    public Queue warningQueue() {
        HashMap<String, Object> map = new HashMap<>(8);
        return new Queue(WARNING_QUEUE_NAME, false, false, false, map);
    }

    @Bean
    public Binding queueConfirmBindingExchange(@Qualifier("confirmQueue") Queue queue,
                                               @Qualifier("confirmExchange") Exchange exchange) {

        return BindingBuilder.bind(queue).to(exchange).with(CONFIRM_ROUTING_KEY).noargs();
    }

    @Bean
    public Binding backupConfirmBindingExchange(@Qualifier("backupQueue") Queue queue,
                                                @Qualifier("backupExchange") FanoutExchange exchange) {

        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public Binding warningConfirmBindingExchange(@Qualifier("warningQueue") Queue queue,
                                                 @Qualifier("backupExchange") FanoutExchange exchange) {

        return BindingBuilder.bind(queue).to(exchange);
    }
}

8.3.3. 报警消费者

package com.xingchen.mq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author xingchen
 * @version V1.0
 * @Package com.xingchen.mq.config
 * @date 2022/12/7 11:52
 */
@Component
@Slf4j
public class WarningConsumer {
    @RabbitListener(queues = {"warning-queue"})
    public void receiveWarningMsg(Message message) {
        log.warn("出现不可路由消息:", message);
    }
}

8.3.4. 测试注意事项

重新启动项目的时候需要把原来的confirm.exchange 删除因为我们修改了其绑定属性，不然报以下错:

8.3.5. 结果分析

mandatory 参数与备份交换机可以一起使用的时候，如果两者同时开启，消息究竟何去何从？谁优先

级高，经过上面结果显示答案是备份交换机优先级高。

相关推荐

RabbitMQ之发布确认（单个发布确认，批量发布确认，异步发布确认）

言午栩 96 0 0

RabbitMQ之发布确认高级

得一道人 25 0 0

RabbitMQ发布确认

Gaaidou 35 0 0

RabbitMQ:发布确认模式

静守幸福 90 0 0

RabbitMQ——发布确认高级

elvinyang 87 0 0

RabbitMQ 发布确认机制

乐百川 62 0 0

RabbitMQ 发布确认模式

罗蓁蓁 30 0 0

RabbitMQ（四）发布确认

得一道人 79 0 0

消息中间件 RabbitMQ 之 发布确认

文风起武 86 0 0

RabbitMQ发布确认高级篇

49路末班车 23 0 0

精彩评论（0）

0 0 举报