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

Springboot整合Rabbitmq之Confirm和Return機(jī)制

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

前言

之前專欄中，對(duì)Springboot整合Rabbitmq都有一系列的配置和說(shuō)明，但總?cè)鄙僖恍┍匾拿枋鲂畔?。?dǎo)致很多看博客的小伙伴會(huì)私信問(wèn)為什么需要這么配置的問(wèn)題。

本篇博客重點(diǎn)進(jìn)行Confirm 機(jī)制和Return 機(jī)制的實(shí)現(xiàn)和說(shuō)明。

為什么會(huì)有Confirm

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

在這里插入圖片描述

每個(gè)Virtual Host 虛擬機(jī)中，都會(huì)含有各自的Exchange和Queue，需要在rabbitmq web界面中針對(duì)可以訪問(wèn)該Virtual Host 虛擬機(jī)的用戶進(jìn)行設(shè)定。

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

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

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

在這里插入圖片描述

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

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

依賴引入

開發(fā)測(cè)試主要的SpringBoot 版本為2.1.4.RELEASE。

此時(shí)只需要引入指定的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>

增加配置文件，設(shè)定連接信息

增加配置文件，配置使用具體的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ā)器確認(rèn)機(jī)制，是都確認(rèn)回調(diào)
    publisher-returns: true

配置隊(duì)列、交換機(jī)，以及對(duì)其進(jìn)行綁定

指定交換機(jī)名稱為：xiangjiao.exchange。
隊(duì)列名稱為：xiangjiao.queue。
使用Direct 直連模式，其中關(guān)聯(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 {
	//隊(duì)列名稱
	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)建隊(duì)列
	@Bean
	public Queue getQueue(){
	    // 另一種方式
		//QueueBuilder.durable(QUEUQ_NAME).build();
		return new Queue(QUEUQ_NAME);
	}
	//實(shí)例化交換機(jī)
	@Bean 
	public DirectExchange getDirectExchange(){
		//DirectExchange(String name, boolean durable, boolean autoDelete)
			
		// 另一種方式：
		//ExchangeBuilder.directExchange(EXCHANGE).durable(true).build();
		/**
		 * 參數(shù)一：交換機(jī)名稱；<br>
		 * 參數(shù)二：是否永久；<br>
		 * 參數(shù)三：是否自動(dòng)刪除；<br>
		 */
		return new DirectExchange(EXCHANGE, true, false);
	//綁定消息隊(duì)列和交換機(jī)
	public Binding bindExchangeAndQueue(DirectExchange exchange,Queue queue){
		// 將 創(chuàng)建的 queue 和 exchange 進(jìn)行綁定
		return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
}

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

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

手動(dòng)定義消息發(fā)送處理類，對(duì)其RabbitTemplate進(jìn)行其他設(shè)置。

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) {
        // 設(shè)置交換機(jī)處理失敗消息的模式     true 表示消息由交換機(jī) 到達(dá)不了隊(duì)列時(shí)，會(huì)將消息重新返回給生產(chǎn)者
        // 如果不設(shè)置這個(gè)指令，則交換機(jī)向隊(duì)列推送消息失敗后，不會(huì)觸發(fā) setReturnCallback
        rabbitTemplate.setMandatory(true);
        //消息消費(fèi)者確認(rèn)收到消息后，手動(dòng)ack回執(zhí)
        rabbitTemplate.setConfirmCallback(this);
        
        // 暫時(shí)關(guān)閉 return 配置
        //rabbitTemplate.setReturnCallback(this);
        //發(fā)送消息
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }
    /**
     * 交換機(jī)并未將數(shù)據(jù)丟入指定的隊(duì)列中時(shí)，觸發(fā)
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  參數(shù)三：true  表示如果消息無(wú)法正常投遞，則return給生產(chǎn)者 ；false 表示直接丟棄
     * @param message   消息對(duì)象
     * @param replyCode 錯(cuò)誤碼
     * @param replyText 錯(cuò)誤信息
     * @param exchange 交換機(jī)
     * @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ā)送消息至交換機(jī)時(shí)觸發(fā)，用于判斷交換機(jī)是否成功收到消息
     * @param correlationData  相關(guān)配置信息
     * @param ack exchange 交換機(jī)，判斷交換機(jī)是否成功收到消息    true 表示交換機(jī)收到
     * @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){
            // 交換機(jī)接收到
            log.info("---- confirm ----ack==true  cause="+cause);
        }else{
            // 沒(méi)有接收到
            log.info("---- confirm ----ack==false  cause="+cause);
        }
}

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

編寫一個(gè)Controller，將產(chǎn)生的數(shù)據(jù)，通過(guò)自定義的RabbitmqService發(fā)送至指定的Exchange交換機(jī)中。

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  交換機(jī)
			// xiangjiao.routingKey  隊(duì)列
			rabbitmqService.sendMessage(MQConfiguration.EXCHANGE, MQConfiguration.ROUTING_KEY, msg);
			//每?jī)擅氚l(fā)送一次
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return "send ok";
	}
}

啟動(dòng)項(xiàng)目進(jìn)行測(cè)試

正常測(cè)試

http://localhost/sendMoreMsgTx

從控制臺(tái)中可以看到消息信息如下所示：

在這里插入圖片描述

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

異常測(cè)試

異常測(cè)試，首先需要保證mq服務(wù)中沒(méi)有對(duì)應(yīng)的exchange交換機(jī)。還需要保證消息的發(fā)送者exchange信息修改。

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

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

在這里插入圖片描述

重啟項(xiàng)目，重新請(qǐng)求該接口，觀察控制臺(tái)數(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)

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

rabbitTemplate.setConfirmCallback(this);

/**
 * 消息生產(chǎn)者發(fā)送消息至交換機(jī)時(shí)觸發(fā)，用于判斷交換機(jī)是否成功收到消息
 * @param correlationData  相關(guān)配置信息
 * @param ack exchange 交換機(jī)，判斷交換機(jī)是否成功收到消息    true 表示交換機(jī)收到
 * @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){
        // 交換機(jī)接收到
        log.info("---- confirm ----ack==true  cause="+cause);
    }else{
        // 沒(méi)有接收到
        log.info("---- confirm ----ack==false  cause="+cause);
    }
}

什么是Return？

上面的配置中，采取Confirm機(jī)制，能夠更好的保證消息生產(chǎn)者確認(rèn)消息是否正常到達(dá)Exchange中。

但是，在MQ中，由于使用Exchange和Queue進(jìn)行了綁定，

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

在這里插入圖片描述


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

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

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

修改 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) {
        // 設(shè)置交換機(jī)處理失敗消息的模式     true 表示消息由交換機(jī) 到達(dá)不了隊(duì)列時(shí)，會(huì)將消息重新返回給生產(chǎn)者
        // 如果不設(shè)置這個(gè)指令，則交換機(jī)向隊(duì)列推送消息失敗后，不會(huì)觸發(fā) setReturnCallback
        rabbitTemplate.setMandatory(true);
        //消息消費(fèi)者確認(rèn)收到消息后，手動(dòng)ack回執(zhí)
        rabbitTemplate.setConfirmCallback(this);

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

    /**
     * 交換機(jī)并未將數(shù)據(jù)丟入指定的隊(duì)列中時(shí)，觸發(fā)
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  參數(shù)三：true  表示如果消息無(wú)法正常投遞，則return給生產(chǎn)者 ；false 表示直接丟棄
     * @param message   消息對(duì)象
     * @param replyCode 錯(cuò)誤碼
     * @param replyText 錯(cuò)誤信息
     * @param exchange 交換機(jī)
     * @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ā)送消息至交換機(jī)時(shí)觸發(fā)，用于判斷交換機(jī)是否成功收到消息
     * @param correlationData  相關(guān)配置信息
     * @param ack exchange 交換機(jī)，判斷交換機(jī)是否成功收到消息    true 表示交換機(jī)收到
     * @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){
            // 交換機(jī)接收到
            log.info("---- confirm ----ack==true  cause="+cause);
        }else{
            // 沒(méi)有接收到
            log.info("---- confirm ----ack==false  cause="+cause);
        }
    }
}

【注意：】設(shè)置 setReturnCallback 后，如果需要保證消息未傳遞到指定的 queue，需要將消息返回生產(chǎn)者時(shí)，一定要增加下面配置：

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

測(cè)試

修改對(duì)應(yīng)的測(cè)試類，保證交換機(jī)正確，但路由key不存在對(duì)應(yīng)的隊(duì)列即可。

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

重啟項(xiàng)目，訪問(wèn)接口，進(jìn)行測(cè)試：

在這里插入圖片描述

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

總結(jié)

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

在這里插入圖片描述

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

相關(guān)代碼下載

gitee 代碼下載地址

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

相關(guān)文章

最新評(píng)論