聊聊RabbitMQ發(fā)布確認高級問題

更新時間：2022年01月04日 11:24:51 作者：崇尚學技術的科班人

這篇文章主要介紹了RabbitMQ發(fā)布確認高級問題，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下

1、發(fā)布確認高級

1. 存在的問題

再生產環(huán)境中由于一些不明原因導致rabbitmq重啟，在RabbitMQ重啟期間生產者消息投遞失敗，會導致消息丟失。

1.1、發(fā)布確認SpringBoot版本

1.1.1、確認機制方案

在這里插入圖片描述

當消息不能正常被接收的時候，我們需要將消息存放在緩存中。

1.1.2、代碼架構圖

在這里插入圖片描述

1.1.3、配置文件

spring.rabbitmq.host=192.168.123.129
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
spring.rabbitmq.publisher-confirm-type=correlated

NONE：禁用發(fā)布確認模式，是默認值。
CORRELATED：發(fā)布消息成功到交換機會觸發(fā)回調方方法。
CORRELATED：就是發(fā)布一個就確認一個。

1.1.4、配置類

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ConfirmConfig {

    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";

    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";

    public static final String CONFIRM_ROUTING_KEY = "key1";

    @Bean("confirmExchange")
    public DirectExchange confirmExchange(){
        return new DirectExchange(CONFIRM_EXCHANGE_NAME);
    }

    @Bean("confirmQueue")
    public Queue confirmQueue(){
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    }

    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmExchange") DirectExchange confirmExchange,
                                        @Qualifier("confirmQueue") Queue confirmQueue){
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
    }
}

1.1.5、回調接口

import lombok.extern.slf4j.Slf4j;
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 javax.annotation.PostConstruct;

/**
 * 回調接口
 */

@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
    }

    /**
     * 交換機接受失敗后進行回調
     * 1. 保存消息的ID及相關消息
     * 2. 是否接收成功
     * 3. 接受失敗的原因
     * @param correlationData
     * @param b
     * @param s
     */

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        String id = correlationData != null ? correlationData.getId() : "";
        if(b == true){
            log.info("交換機已經收到id為：{}的消息",id);
        }else{
            log.info("交換機還未收到id為：{}消息，由于原因：{}",id,s);
        }
    }
}

1.1.6、生產者

import com.xiao.springbootrabbitmq.utils.MyCallBack;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;

@RestController
@RequestMapping("/confirm")
@Slf4j
public class Producer {
    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";

    @Autowired
    RabbitTemplate rabbitTemplate;



    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message){
        CorrelationData correlationData1 = new CorrelationData("1");
        String routingKey1 = "key1";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey1,message + routingKey1,correlationData1);

        CorrelationData correlationData2 = new CorrelationData("2");
        String routingKey2 = "key2";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey2,message + routingKey2,correlationData2);
        log.info("發(fā)送得內容是：{}",message);
    }
}

1.1.7、消費者

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class ConfirmConsumer {
    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";

    @RabbitListener(queues = CONFIRM_QUEUE_NAME)
    public void receiveMessage(Message message){
        String msg = new String(message.getBody());
        log.info("接收到隊列" + CONFIRM_QUEUE_NAME + "消息：{}",msg);
    }
}

1.1.8、測試結果

1. 第一種情況

在這里插入圖片描述

ID為1的消息正常送達，ID為2的消息由于RoutingKey的錯誤，導致不能正常被消費，但是交換機還是正常收到了消息，所以此時由于交換機正常接收之后的原因丟失的消息不能正常被接收。

2. 第二種情況

在這里插入圖片描述

我們再上一種情況下修改了ID為1的消息的交換機的名稱，所以此時回調函數會進行回答由于什么原因導致交換機無法接收成功消息。

1.2、回退消息

1.2.1、Mandatory參數

在僅開啟了生產者確認機制的情況下，交換機接收到消息后，會直接給消息生產者發(fā)送確認消息，如果發(fā)現該消息不可路由（就是消息被交換機成功接收后，無法到達隊列），那么消息會直接被丟棄，此時生產者是不知道消息被丟棄這個事件的。
通過設置該參數可以在消息傳遞過程中不可達目的地時將消息返回給生產者。

1.2.2、配置文件

spring.rabbitmq.publisher-returns=true

需要在配置文件種開啟返回回調

1.2.3、生產者代碼

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;

@RestController
@RequestMapping("/confirm")
@Slf4j
public class Producer {
    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";

    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message){
        CorrelationData correlationData1 = new CorrelationData("1");
        String routingKey1 = "key1";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey1,message + routingKey1,correlationData1);
        log.info("發(fā)送得內容是：{}",message + routingKey1);

        CorrelationData correlationData2 = new CorrelationData("2");
        String routingKey2 = "key2";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey2,message + routingKey2,correlationData2);
        log.info("發(fā)送得內容是：{}",message + routingKey2);
    }
}

1.2.4、回調接口代碼

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
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 javax.annotation.PostConstruct;

/**
 * 回調接口
 */

@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * 交換機接受失敗后進行回調
     * 1. 保存消息的ID及相關消息
     * 2. 是否接收成功
     * 3. 接受失敗的原因
     * @param correlationData
     * @param b
     * @param s
     */

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        String id = correlationData != null ? correlationData.getId() : "";
        if(b == true){
            log.info("交換機已經收到id為：{}的消息",id);
        }else{
            log.info("交換機還未收到id為：{}消息，由于原因：{}",id,s);
        }
    }


    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        Message message = returnedMessage.getMessage();
        String exchange = returnedMessage.getExchange();
        String routingKey = returnedMessage.getRoutingKey();
        String replyText = returnedMessage.getReplyText();
        log.error("消息{}，被交換機{}退回，回退原因：{}，路由Key：{}",new String(message.getBody()),exchange,replyText,routingKey);
    }
}

1.2.5、測試結果

其他類的代碼與上一小節(jié)案例相同

在這里插入圖片描述

ID為2的消息由于RoutingKey不可路由，但是還是被回調函數處理了。

1.3、備份交換機

1.3.1、代碼架構圖

在這里插入圖片描述

這里我們新增了備份交換機、備份隊列、報警隊列。它們綁定關系如圖所示。如果確認交換機成功接收的消息無法路由到相應的隊列，就會被確認交換機發(fā)送給備份交換機。

1.3.2、配置類代碼

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ConfirmConfig {

    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";

    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";

    public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";

    public static final String BACKUP_QUEUE_NAME = "backup_queue";

    public static final String WARNING_QUEUE_NAME = "warning_queue";

    public static final String CONFIRM_ROUTING_KEY = "key1";

    @Bean("confirmExchange")
    public DirectExchange confirmExchange(){
        return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
                .withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
    }

    @Bean("confirmQueue")
    public Queue confirmQueue(){
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    }

    @Bean("backupExchange")
    public FanoutExchange backupExchange(){
        return new FanoutExchange(BACKUP_EXCHANGE_NAME);
    }

    @Bean("backupQueue")
    public Queue backupQueue(){
        return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
    }

    @Bean("warningQueue")
    public Queue warningQueue(){
        return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
    }

    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmExchange") DirectExchange confirmExchange,
                                        @Qualifier("confirmQueue") Queue confirmQueue){
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
    }

    @Bean
    public Binding queueBindingExchange1(@Qualifier("backupExchange") FanoutExchange backupExchange,
                                        @Qualifier("backupQueue") Queue backupQueue){
        return BindingBuilder.bind(backupQueue).to(backupExchange);
    }

    @Bean
    public Binding queueBindingExchange2(@Qualifier("backupExchange") FanoutExchange backupExchange,
                                         @Qualifier("warningQueue") Queue warningQueue){
        return BindingBuilder.bind(warningQueue).to(backupExchange);
    }
}

1.3.3、消費者代碼

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class WarningConsumer {
    public static final String WARNING_QUEUE_NAME = "warning_queue";

    @RabbitListener(queues = WARNING_QUEUE_NAME)
    public void receiveMessage(Message message){
        String msg = new String(message.getBody());
        log.info("報警發(fā)現不可路由的消息內容為：{}",msg);
    }
}

1.3.4、測試結果

在這里插入圖片描述

mandatory參數與備份交換機可以一起使用的時候，如果兩者同時開啟，備份交換機優(yōu)先級高。

到此這篇關于RabbitMQ發(fā)布確認高級的文章就介紹到這了,更多相關RabbitMQ發(fā)布確認高級內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

java數據庫開發(fā)之JDBC基礎使用方法及實例詳解
這篇文章主要介紹了java數據庫開發(fā)之JDBC基礎知識詳解,需要的朋友可以參考下
2020-02-02
詳解Java中二叉樹的基礎概念(遞歸&迭代)
二叉樹（Binary?tree）是樹形結構的一個重要類型。本文將通過圖片和示例代碼詳細為大家講解一下Java中的二叉樹的基礎概念，需要的可以參考一下
2022-03-03
java算法題解LeetCode35復雜鏈表的復制實例
這篇文章主要為大家介紹了java算法題解LeetCode35復雜鏈表的復制實例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2023-01-01
java實現文件打包壓縮輸出到瀏覽器下載
這篇文章主要介紹了java實現文件打包壓縮輸出到瀏覽器下載，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2021-11-11
Maven 多模塊父子工程的實現(含Spring Boot示例)
這篇文章主要介紹了Maven 多模塊父子工程的實現(含Spring Boot示例)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2021-04-04
mybatis批量插入,批量更新以及null值問題的解決
這篇文章主要介紹了mybatis批量插入,批量更新以及null值問題的解決,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
2024-01-01
SpringBoot升級3.2報錯Invalid value type for
這篇文章給大家介紹了SpringBoot升級3.2報錯Invalid value type for attribute ‘factoryBeanObjectType‘: java.lang.String的解決方案,文中有詳細的原因分析,需要的朋友可以參考下
2023-12-12
解析SpringBoot項目開發(fā)之Gzip壓縮過程
這篇文章主要介紹了SpringBoot項目開發(fā)之Gzip壓縮過程，本文給大家分享幾種Gzip壓縮方式，通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下
2020-07-07
詳細解讀Java的Lambda表達式
這篇文章主要介紹了詳細解讀Java的Lambda表達式，lambda?表達式?是Java?8新加入的新特性，它在Java中是引入了函數式編程這一概念,需要的朋友可以參考下
2023-04-04
java實現大數加法(BigDecimal)的實例代碼
之前寫過用vector、string實現大數加法，現在用java的BigDecimal類，代碼簡單很多。但是在online-judge上，java的代碼運行時間和內存大得多
2013-10-10

亚洲乱码中文字幕综合,中国熟女仑乱hd,亚洲精品乱拍国产一区二区三区,一本大道卡一卡二卡三乱码全集资源,又粗又黄又硬又爽的免费视频

軟件下載

源碼下載

軟件編程

網絡編程

在線工具

數據庫

CMS

常用工具

聊聊RabbitMQ發(fā)布確認高級問題

目錄

1、發(fā)布確認高級

1.1、發(fā)布確認SpringBoot版本

1.1.1、確認機制方案

1.1.2、代碼架構圖

1.1.3、配置文件

1.1.4、配置類

1.1.5、回調接口

1.1.6、生產者

1.1.7、消費者

1.1.8、測試結果

1.2、回退消息

1.2.1、Mandatory參數

1.2.2、配置文件

1.2.3、生產者代碼

1.2.4、回調接口代碼

1.2.5、測試結果

1.3、備份交換機

1.3.1、代碼架構圖

1.3.2、配置類代碼

1.3.3、消費者代碼

1.3.4、測試結果

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

聊聊RabbitMQ發(fā)布確認高級問題

目錄

1、發(fā)布確認高級

1.1、發(fā)布確認SpringBoot版本

1.1.1、確認機制方案

1.1.2、代碼架構圖

1.1.3、配置文件

1.1.4、配置類

1.1.5、回調接口

1.1.6、生產者

1.1.7、消費者

1.1.8、測試結果

1.2、回退消息

1.2.1、Mandatory參數

1.2.2、配置文件

1.2.3、生產者代碼

1.2.4、回調接口代碼

1.2.5、測試結果

1.3、備份交換機

1.3.1、代碼架構圖

1.3.2、配置類代碼

1.3.3、消費者代碼

1.3.4、測試結果

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

1.1、發(fā)布確認SpringBoot版本

1.1.1、確認機制方案

1.1.2、代碼架構圖

1.1.3、配置文件

1.1.4、配置類

1.1.6、生產者

1.1.7、消費者

1.1.8、測試結果

1.2.1、Mandatory參數

1.2.3、生產者代碼

1.2.5、測試結果

1.3、備份交換機

1.3.1、代碼架構圖

1.3.2、配置類代碼

1.3.3、消費者代碼

1.3.4、測試結果