RabbitMQ confirm 模式之異步確認

channel 的 confirm 模式是 RabbitMQ 爲保障消息的可靠性投遞，所實現的一種機制。

關於channel 的 confirm 模式以及 RabbitMQ 如何保障消息的可靠性投遞，請參考 RabbitMQ 之消息的可靠性投遞

這裏只介紹 confirm 模式中的異步確認

異步確認

異步確認是 confirm 模式的一種使用方式，因爲它是異步、非阻塞的，所以性能比較好，實際開發中比較經常使用。

本文主要介紹 confirm 模式下異步確認如何實現，廢話很少說，直接看代碼實現

項目地址

pom

<dependencies>
    <!-- 引入RabbitMq 依賴 -->
    <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);
    }
複製代碼