Springboot整合Rabbitmq之Confirm和Return機制

Posted on 2022-02-28 by WalkonNet

前言

之前專欄中，對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);

重啟項目，訪問接口，進行測試：

在這裡插入圖片描述

消息發送給Exchange成功，但是通過Exchange向Queue中推送數據時失敗，經過ReturnCallback 的 returnedMessage捕獲監聽！

總結

通過配置ConfirmCallback和ReturnCallback，便能實現消息生產者到交換機和消息由exchange到queue這個鏈路的安全性！

在這裡插入圖片描述

都是出現問題，或者正常後，給生產者方進行反饋。

Springboot整合Rabbitmq之Confirm和Return機制

目錄

前言

為什麼會有Confirm

Springboot 整合 Mq 實現 Confirm 監聽機制

依賴引入

增加配置文件，設定連接信息

配置隊列、交換機，以及對其進行綁定

編寫mq消息發送服務

編寫消息發送接口

啟動項目進行測試

正常測試

異常測試

什麼是Return？

增加 ReturnCallback 監聽並測試

修改 RabbitmqService 配置類

測試

總結

相關代碼下載

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

前言

為什麼會有Confirm

Springboot 整合 Mq 實現 Confirm 監聽機制

依賴引入

增加配置文件，設定連接信息

配置隊列、交換機，以及對其進行綁定

編寫mq消息發送服務

編寫消息發送接口

啟動項目進行測試

正常測試

異常測試

什麼是Return？

增加 ReturnCallback 監聽並測試

修改 RabbitmqService 配置類

測試

總結

相關代碼下載

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆