Springboot整合Rabbitmq之Confirm和Return機制

2022-02-28 16:06:33

前言

之前專欄中，對Springboot整合Rabbitmq都有一系列的設定和說明，但總缺少一些必要的描述資訊。導致很多看部落格的小夥伴會私信問為什麼需要這麼設定的問題。

本篇部落格重點進行Confirm 機制和Return 機制的實現和說明。

為什麼會有Confirm

RabbitMq中，針對資料由訊息生產者向訊息佇列推播時，通常情況如下所示(以 Routing 方式為例)：

每個Virtual Host 虛擬機器器中，都會含有各自的Exchange和Queue，需要在rabbitmq web介面中針對可以存取該Virtual Host 虛擬機器器的使用者進行設定。

有點類似資料庫的概念，指定使用者只能操作指定的資料庫。

在使用交換機 Exchange時，訊息生產者需要將訊息通過Channel 管道將資料傳送給MQ，但想過一個問題沒有：

如何確定訊息是否真的傳送到了指定的 MQ 中呢？

MQ中，對此問題，提出有Confirm 機制，對其傳送資料進行監聽，讓訊息傳送者知道訊息的傳送結果。

Springboot 整合 Mq 實現 Confirm 監聽機制

依賴引入

開發測試主要的SpringBoot 版本為2.1.4.RELEASE。

此時只需要引入指定的amqp依賴即可：

<dependency>
  <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

完整的pom依賴如下所示：

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>1.0-SNAPSHOT</version>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
        <relativePath /> <!-- lookup parent from repository -->
    </parent>
    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <!-- 引入rabbitmq依賴 -->
            <artifactId>spring-boot-starter-amqp</artifactId>
            <artifactId>spring-boot-starter-web</artifactId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.20</version>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.26</version>
            <artifactId>slf4j-log4j12</artifactId>
    </dependencies>
</project>

增加組態檔，設定連線資訊

增加組態檔，設定使用具體的Virtual Host、Username、Password、Host、Port等資訊。

server:
  port: 80
spring:
  rabbitmq:
    host: xxxxxx
    port: 5672
    username: xiangjiao
    password: bunana
    virtual-host: /xiangjiao
    publisher-confirms: true   #訊息傳送到轉發器確認機制，是都確認回撥
    publisher-returns: true

設定佇列、交換機，以及對其進行繫結

指定交換機名稱為：xiangjiao.exchange。
佇列名稱為：xiangjiao.queue。
使用Direct 直連模式，其中關聯的Routingkey為：xiangjiao.routingKey。

package cn.linkpower.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MQConfiguration {
	//佇列名稱
	public static final String QUEUQ_NAME = "xiangjiao.queue";
	//交換器名稱
	public static final String EXCHANGE = "xiangjiao.exchange";
	//路由key
	public static final String ROUTING_KEY = "xiangjiao.routingKey";
	
	//建立佇列
	@Bean
	public Queue getQueue(){
	    // 另一種方式
		//QueueBuilder.durable(QUEUQ_NAME).build();
		return new Queue(QUEUQ_NAME);
	}
	//範例化交換機
	@Bean 
	public DirectExchange getDirectExchange(){
		//DirectExchange(String name, boolean durable, boolean autoDelete)
			
		// 另一種方式：
		//ExchangeBuilder.directExchange(EXCHANGE).durable(true).build();
		/**
		 * 引數一：交換機名稱；<br>
		 * 引數二：是否永久；<br>
		 * 引數三：是否自動刪除；<br>
		 */
		return new DirectExchange(EXCHANGE, true, false);
	//繫結訊息佇列和交換機
	public Binding bindExchangeAndQueue(DirectExchange exchange,Queue queue){
		// 將 建立的 queue 和 exchange 進行繫結
		return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
}

編寫mq訊息傳送服務

在Springboot中，針對MQ訊息的傳送，採取RabbitTemplate模板進行資料的傳送處理操作。

手動定義訊息傳送處理類，對其RabbitTemplate進行其他設定。

package cn.linkpower.service;

import lombok.extern.slf4j.Slf4j;
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;
@Slf4j
@Component
public class RabbitmqService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String exchange,String routingKey,Object msg) {
        // 設定交換機處理失敗訊息的模式     true 表示訊息由交換機 到達不了佇列時，會將訊息重新返回給生產者
        // 如果不設定這個指令，則交換機向佇列推播訊息失敗後，不會觸發 setReturnCallback
        rabbitTemplate.setMandatory(true);
        //訊息消費者確認收到訊息後，手動ack回執
        rabbitTemplate.setConfirmCallback(this);
        
        // 暫時關閉 return 設定
        //rabbitTemplate.setReturnCallback(this);
        //傳送訊息
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }
    /**
     * 交換機並未將資料丟入指定的佇列中時，觸發
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  引數三：true  表示如果訊息無法正常投遞，則return給生產者 ；false 表示直接丟棄
     * @param message   訊息物件
     * @param replyCode 錯誤碼
     * @param replyText 錯誤資訊
     * @param exchange 交換機
     * @param routingKey 路由鍵
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");
     * 訊息生產者傳送訊息至交換機時觸發，用於判斷交換機是否成功收到訊息
     * @param correlationData  相關設定資訊
     * @param ack exchange 交換機，判斷交換機是否成功收到訊息    true 表示交換機收到
     * @param cause  失敗原因
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("---- confirm ----ack="+ack+"  cause="+String.valueOf(cause));
        log.info("correlationData -->"+correlationData.toString());
        if(ack){
            // 交換機接收到
            log.info("---- confirm ----ack==true  cause="+cause);
        }else{
            // 沒有接收到
            log.info("---- confirm ----ack==false  cause="+cause);
        }
}

編寫訊息傳送介面

編寫一個Controller，將產生的資料，通過自定義的RabbitmqService傳送至指定的Exchange交換機中。

package cn.linkpower.controller;
import cn.linkpower.config.MQConfiguration;
import cn.linkpower.service.RabbitmqService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
@Controller
public class SendMessageTx {
	
	@Autowired
	private RabbitmqService rabbitmqService;
	@RequestMapping("/sendMoreMsgTx")
	@ResponseBody
	public String sendMoreMsgTx(){
		//傳送10條訊息
		for (int i = 0; i < 10; i++) {
			String msg = "msg"+i;
			System.out.println("傳送訊息  msg："+msg);
			// xiangjiao.exchange  交換機
			// xiangjiao.routingKey  佇列
			rabbitmqService.sendMessage(MQConfiguration.EXCHANGE, MQConfiguration.ROUTING_KEY, msg);
			//每兩秒傳送一次
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return "send ok";
	}
}

啟動專案進行測試

正常測試

http://localhost/sendMoreMsgTx

從控制檯中可以看到訊息資訊如下所示：

發現，訊息資訊傳送，都是ACK 被確認的！

異常測試

異常測試，首先需要保證mq服務中沒有對應的exchange交換機。還需要保證訊息的傳送者exchange資訊修改。

將controller中對應的訊息傳送的方式修改如下：

rabbitmqService.sendMessage("xiangjiao.exchangeError", MQConfiguration.ROUTING_KEY, msg);

重啟專案，重新請求該介面，觀察控制檯資料資訊展示：

擷取其中的一條資訊為例：

傳送訊息 msg：msg0
2022-02-28 10:34:58.686 ---- [rabbitConnectionFactory1] ---- INFO cn.linkpower.service.RabbitmqService - ---- confirm ----ack=false
cause=channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND -
no exchange 'xiangjiao.exchangeError' in vhost '/xiangjiao', class-id=60, method-id=40)

當生產者向Exchange中傳送訊息，如果訊息並未成功傳送，則會觸發RabbitmqService中設定的confirm處理機制。

rabbitTemplate.setConfirmCallback(this);

/**
 * 訊息生產者傳送訊息至交換機時觸發，用於判斷交換機是否成功收到訊息
 * @param correlationData  相關設定資訊
 * @param ack exchange 交換機，判斷交換機是否成功收到訊息    true 表示交換機收到
 * @param cause  失敗原因
 */
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    log.info("---- confirm ----ack="+ack+"  cause="+String.valueOf(cause));
    log.info("correlationData -->"+correlationData.toString());
    if(ack){
        // 交換機接收到
        log.info("---- confirm ----ack==true  cause="+cause);
    }else{
        // 沒有接收到
        log.info("---- confirm ----ack==false  cause="+cause);
    }
}

什麼是Return？

上面的設定中，採取Confirm機制，能夠更好的保證訊息生產者確認訊息是否正常到達Exchange中。

但是，在MQ中，由於使用Exchange和Queue進行了繫結，

如果某個佇列宕機了，Exchange並未將訊息傳送匹配 Routing Key 的佇列，那麼訊息就不能到達佇列中！！！

mq中，對此情況設有另外一種監聽機制：Return機制！

當訊息由Exchange 未能傳遞到匹配的 queue 中，則會通過ReturnCallback根據使用者的抉擇，判斷是否需要返回給訊息生產者。

增加 ReturnCallback 監聽並測試

修改 RabbitmqService 設定類

package cn.linkpower.service;


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.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class RabbitmqService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String exchange,String routingKey,Object msg) {
        // 設定交換機處理失敗訊息的模式     true 表示訊息由交換機 到達不了佇列時，會將訊息重新返回給生產者
        // 如果不設定這個指令，則交換機向佇列推播訊息失敗後，不會觸發 setReturnCallback
        rabbitTemplate.setMandatory(true);
        //訊息消費者確認收到訊息後，手動ack回執
        rabbitTemplate.setConfirmCallback(this);

        // return 設定
        rabbitTemplate.setReturnCallback(this);
        //傳送訊息
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }

    /**
     * 交換機並未將資料丟入指定的佇列中時，觸發
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  引數三：true  表示如果訊息無法正常投遞，則return給生產者 ；false 表示直接丟棄
     * @param message   訊息物件
     * @param replyCode 錯誤碼
     * @param replyText 錯誤資訊
     * @param exchange 交換機
     * @param routingKey 路由鍵
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");
    }

    /**
     * 訊息生產者傳送訊息至交換機時觸發，用於判斷交換機是否成功收到訊息
     * @param correlationData  相關設定資訊
     * @param ack exchange 交換機，判斷交換機是否成功收到訊息    true 表示交換機收到
     * @param cause  失敗原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("---- confirm ----ack="+ack+"  cause="+String.valueOf(cause));
        log.info("correlationData -->"+correlationData.toString());
        if(ack){
            // 交換機接收到
            log.info("---- confirm ----ack==true  cause="+cause);
        }else{
            // 沒有接收到
            log.info("---- confirm ----ack==false  cause="+cause);
        }
    }
}

【注意：】設定 setReturnCallback 後，如果需要保證訊息未傳遞到指定的 queue，需要將訊息返回生產者時，一定要增加下面設定：

// 設定交換機處理失敗訊息的模式     true 表示訊息由交換機 到達不了佇列時，會將訊息重新返回給生產者
// 如果不設定這個指令，則交換機向佇列推播訊息失敗後，不會觸發 setReturnCallback
rabbitTemplate.setMandatory(true);

測試

修改對應的測試類，保證交換機正確，但路由key不存在對應的佇列即可。

// xiangjiao.routingKey 存在對應的queue
// xiangjiao.routingKey_error 不存在對應的 queue
rabbitmqService.sendMessage(MQConfiguration.EXCHANGE, "xiangjiao.routingKey_error", msg);

重啟專案，存取介面，進行測試：