欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Springboot整合Rabbitmq之Confirm和Return機制

 更新時間：2022年02月28日 14:42:11   作者：專注寫bug  
這篇文章主要介紹了Springboot整合Rabbitmq之Confirm和Return詳解，本篇重點進行Confirm?機制和Return?機制的實現(xiàn)和說明,通過實例代碼相結(jié)合給大家詳細介紹，對Springboot整合Rabbitmq相關知識感興趣的朋友一起看看吧

前言

之前專欄中，對Springboot整合Rabbitmq都有一系列的配置和說明，但總?cè)鄙僖恍┍匾拿枋鲂畔ⅰе潞芏嗫床┛偷男』锇闀叫艈枮槭裁葱枰@么配置的問題。

本篇博客重點進行Confirm 機制和Return 機制的實現(xiàn)和說明。

為什么會有Confirm

RabbitMq中，針對數(shù)據(jù)由消息生產(chǎn)者向消息隊列推送時，通常情況如下所示(以 Routing 方式為例)：

在這里插入圖片描述

每個Virtual Host 虛擬機中，都會含有各自的Exchange和Queue，需要在rabbitmq web界面中針對可以訪問該Virtual Host 虛擬機的用戶進行設定。

有點類似數(shù)據(jù)庫的概念，指定用戶只能操作指定的數(shù)據(jù)庫。

在使用交換機 Exchange時，消息生產(chǎn)者需要將消息通過Channel 管道將數(shù)據(jù)發(fā)送給MQ，但想過一個問題沒有：

如何 確定 消息是否真的發(fā)送到了指定的 MQ 中呢？

在這里插入圖片描述

MQ中，對此問題，提出有Confirm 機制，對其發(fā)送數(shù)據(jù)進行監(jiān)聽，讓消息發(fā)送者知道消息的發(fā)送結(jié)果。

Springboot 整合 Mq 實現(xiàn) Confirm 監(jiān)聽機制

依賴引入

開發(fā)測試主要的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   #消息發(fā)送到轉(zhuǎn)發(fā)器確認機制，是都確認回調(diào)
    publisher-returns: true

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

指定交換機名稱為：xiangjiao.exchange。
隊列名稱為：xiangjiao.queue。
使用Direct 直連模式，其中關聯(lián)的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";
	
	//創(chuàng)建隊列
	@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();
		/**
		 * 參數(shù)一：交換機名稱；<br>
		 * 參數(shù)二：是否永久；<br>
		 * 參數(shù)三：是否自動刪除；<br>
		 */
		return new DirectExchange(EXCHANGE, true, false);
	//綁定消息隊列和交換機
	public Binding bindExchangeAndQueue(DirectExchange exchange,Queue queue){
		// 將 創(chuàng)建的 queue 和 exchange 進行綁定
		return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
}

編寫mq消息發(fā)送服務

在Springboot中，針對MQ消息的發(fā)送，采取RabbitTemplate模板進行數(shù)據(jù)的發(fā)送處理操作。

手動定義消息發(fā)送處理類，對其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 表示消息由交換機 到達不了隊列時，會將消息重新返回給生產(chǎn)者
        // 如果不設置這個指令，則交換機向隊列推送消息失敗后，不會觸發(fā) setReturnCallback
        rabbitTemplate.setMandatory(true);
        //消息消費者確認收到消息后，手動ack回執(zhí)
        rabbitTemplate.setConfirmCallback(this);
        
        // 暫時關閉 return 配置
        //rabbitTemplate.setReturnCallback(this);
        //發(fā)送消息
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }
    /**
     * 交換機并未將數(shù)據(jù)丟入指定的隊列中時，觸發(fā)
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  參數(shù)三：true  表示如果消息無法正常投遞，則return給生產(chǎn)者 ；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+" ");
     * 消息生產(chǎn)者發(fā)送消息至交換機時觸發(fā)，用于判斷交換機是否成功收到消息
     * @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);
        }
}

編寫消息發(fā)送接口

編寫一個Controller，將產(chǎn)生的數(shù)據(jù)，通過自定義的RabbitmqService發(fā)送至指定的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(){
		//發(fā)送10條消息
		for (int i = 0; i < 10; i++) {
			String msg = "msg"+i;
			System.out.println("發(fā)送消息  msg："+msg);
			// xiangjiao.exchange  交換機
			// xiangjiao.routingKey  隊列
			rabbitmqService.sendMessage(MQConfiguration.EXCHANGE, MQConfiguration.ROUTING_KEY, msg);
			//每兩秒發(fā)送一次
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return "send ok";
	}
}

啟動項目進行測試

正常測試

http://localhost/sendMoreMsgTx

從控制臺中可以看到消息信息如下所示：

在這里插入圖片描述

發(fā)現(xiàn)，消息信息發(fā)送，都是ACK 被確認的！

異常測試

異常測試，首先需要保證mq服務中沒有對應的exchange交換機。還需要保證消息的發(fā)送者exchange信息修改。

將controller中對應的消息發(fā)送的方式修改如下：

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

在這里插入圖片描述

重啟項目，重新請求該接口，觀察控制臺數(shù)據(jù)信息展示：

在這里插入圖片描述

截取其中的一條信息為例：

發(fā)送消息  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)

當生產(chǎn)者向Exchange中發(fā)送消息，如果消息并未成功發(fā)送，則會觸發(fā)RabbitmqService中設定的confirm處理機制。

rabbitTemplate.setConfirmCallback(this);

/**
 * 消息生產(chǎn)者發(fā)送消息至交換機時觸發(fā)，用于判斷交換機是否成功收到消息
 * @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機制，能夠更好的保證消息生產(chǎn)者確認消息是否正常到達Exchange中。

但是，在MQ中，由于使用Exchange和Queue進行了綁定，

如果某個隊列宕機了，Exchange并未將消息發(fā)送匹配 Routing Key 的隊列，那么消息就不能到達隊列中！??！

在這里插入圖片描述


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

當消息由Exchange 未能傳遞到匹配的 queue 中，則會通過ReturnCallback根據(jù)用戶的抉擇，判斷是否需要返回給消息生產(chǎn)者。

增加 ReturnCallback 監(jiān)聽并測試

修改 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 表示消息由交換機 到達不了隊列時，會將消息重新返回給生產(chǎn)者
        // 如果不設置這個指令，則交換機向隊列推送消息失敗后，不會觸發(fā) setReturnCallback
        rabbitTemplate.setMandatory(true);
        //消息消費者確認收到消息后，手動ack回執(zhí)
        rabbitTemplate.setConfirmCallback(this);

        // return 配置
        rabbitTemplate.setReturnCallback(this);
        //發(fā)送消息
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }

    /**
     * 交換機并未將數(shù)據(jù)丟入指定的隊列中時，觸發(fā)
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  參數(shù)三：true  表示如果消息無法正常投遞，則return給生產(chǎn)者 ；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+" ");
    }

    /**
     * 消息生產(chǎn)者發(fā)送消息至交換機時觸發(fā)，用于判斷交換機是否成功收到消息
     * @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，需要將消息返回生產(chǎn)者時，一定要增加下面配置：

// 設置交換機處理失敗消息的模式     true 表示消息由交換機 到達不了隊列時，會將消息重新返回給生產(chǎn)者
// 如果不設置這個指令，則交換機向隊列推送消息失敗后，不會觸發(fā) setReturnCallback
rabbitTemplate.setMandatory(true);

測試

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

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

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

在這里插入圖片描述

消息發(fā)送給Exchange成功，但是通過Exchange向Queue中推送數(shù)據(jù)時 失敗，經(jīng)過ReturnCallback 的 returnedMessage捕獲監(jiān)聽！

總結(jié)

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

在這里插入圖片描述

都是出現(xiàn)問題，或者正常后，給生產(chǎn)者方進行反饋。

相關代碼下載

gitee 代碼下載地址

到此這篇關于Springboot整合Rabbitmq之Confirm和Return詳解的文章就介紹到這了,更多相關Springboot整合Rabbitmq內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

相關文章

最新評論