RabbitMQ confirm 模式之异步确认

落日映苍穹つ 2023-08-17 16:51 48阅读 0赞

**channel 的 confirm 模式**是 RabbitMQ 为保障消息的可靠性投递，所实现的一种机制。

> 关于channel 的 confirm 模式以及 RabbitMQ 如何保障消息的可靠性投递，请参考 [RabbitMQ 之消息的可靠性投递][RabbitMQ]

这里只介绍 confirm 模式中的异步确认

## 异步确认 ##

异步确认是 confirm 模式的一种使用方式，由于它是异步、非阻塞的，因此性能比较好，实际开发中比较常用。

本文主要介绍 confirm 模式下异步确认如何实现，废话不多说，直接看代码实现

[项目地址][Link 1]

## pom ##

<dependencies>
        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.31</version>
        </dependency>
    
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    复制代码

spring-boot-starter-web 用于测试消息异步确认

## application.properties ##

# RabbitMQ 基本配置
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    
    # 监听消息是否 到达 exchange
    spring.rabbitmq.publisher-confirms=true
    # 监听消息是否 没有到达 queue
    spring.rabbitmq.publisher-returns=true
    # 自动删除不可达消息，默认为false
    spring.rabbitmq.template.mandatory=true
    复制代码

## 消息实体 ##

import java.io.Serializable;
    
    /**
     * 订单的消息实体
     */
    public class OrderMessage implements Serializable {
    
        /**
         * 业务id，在业务系统中的唯一。比如 订单id、支付id、商品id ，消息消费端可以通过该 id 避免消息重复消费
         */
        private String id;
    
        // 其他业务字段
        private String name;
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    }
    复制代码

## 发送消息并异步监听 ##

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    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 java.util.UUID;
    
    
    /**
     * 发送消息并异步监听 ack
     */
    @Component
    public class OrderMessageSendAsync implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    
        private Logger logger = LoggerFactory.getLogger(OrderMessageSendAsync.class);
    
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 通过构造函数注入 RabbitTemplate 依赖
         *
         * @param rabbitTemplate
         */
        @Autowired
        public OrderMessageSendAsync(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
            // 设置消息到达 exchange 时，要回调的方法，每个 RabbitTemplate 只支持一个 ConfirmCallback
            rabbitTemplate.setConfirmCallback(this);
            // 设置消息无法到达 queue 时，要回调的方法
            rabbitTemplate.setReturnCallback(this);
        }
    
        /**
         * 发送消息
         *
         * @param exchange   交换机
         * @param routingKey 路由建
         * @param message    消息实体
         */
        public void sendMsg(String exchange, String routingKey, Object message) {
            // 构造包含消息的唯一id的对象，id 必须在该 channel 中始终唯一
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            logger.info("ID为: {}", correlationData.getId());
            // todo 先将 message 入库，在将 message 的数据库ID 、message的消息id message的初始状态(发送中)等信息入库
    
            // 完成 数据落库，消息状态打标后，就可以安心发送 message
            rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
    
            try {
                logger.info("发送消息的线程处于休眠状态， confirm 和 returnedMessage 方法依然处于异步监听状态");
                Thread.sleep(1000*15);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
        }
    
    
        /**
         * 异步监听 消息是否到达 exchange
         *
         * @param correlationData 包含消息的唯一标识的对象
         * @param ack             true 标识 ack，false 标识 nack
         * @param cause           nack 的原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    
            if (ack) {
                logger.info("消息投递成功,ID为: {}", correlationData.getId());
                // todo 操作数据库，将 correlationId 这条消息状态改为投递成功
                return;
            }
    
            logger.error("消息投递失败,ID为: {},错误信息: {}", correlationData.getId(), cause);
            // todo 操作数据库，将 correlationId 这条消息状态改为投递失败
    
        }
    
        /**
         * 异步监听 消息是否到达 queue
         * 触发回调的条件有两个：1.消息已经到达了 exchange 2.消息无法到达 queue (比如 exchange 找不到跟 routingKey 对应的 queue)
         *
         * @param message    返回的消息
         * @param replyCode  回复 code
         * @param replyText  回复 text
         * @param exchange   交换机
         * @param routingKey 路由键
         */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            // correlationId 就是发消息时设置的 id
            String correlationId = message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
    
            logger.error("没有找到对应队列，消息投递失败,ID为: {}, replyCode {} , replyText {}, exchange {} routingKey {}",
                    correlationId, replyCode, replyText, exchange, routingKey);
            // todo 操作数据库，将 correlationId 这条消息状态改为投递失败
        }
    }
    复制代码

## 测试 ##

import com.alibaba.fastjson.JSONObject;
    import com.wqlm.rabbitmq.send.MessageSend.OrderMessageSendAsync;
    import com.wqlm.rabbitmq.send.message.OrderMessage;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    @RequestMapping("/")
    public class SendMessageController {
    
        @Autowired
        private OrderMessageSendAsync orderMessageSendAsync;
    
        /**
         * 测试发送消息并异步接收响应
         */
        @GetMapping("/test")
        public void test(){
            OrderMessage orderMessage = new OrderMessage("123", "订单123");
            // 序列化成json ，OrderMessage 也可以 implements Serializable 这样就不需要序列化成json
            String message = JSONObject.toJSONString(orderMessage);
            orderMessageSendAsync.sendMsg("exchangeName", "routingKeyValue", message);
        }

[RabbitMQ]: https://juejin.im/post/5d6cccd85188251ff57f9e2a
[Link 1]: https://link.juejin.im?target=https%3A%2F%2Fgithub.com%2Fwqlm%2Frabbitmq-send