聊聊RabbitMQ發(fā)布確認高級問題
更新時間：2022年01月04日 11:24:51 作者：崇尚學(xué)技術(shù)的科班人
這篇文章主要介紹了RabbitMQ發(fā)布確認高級問題，本文給大家介紹的非常詳細，對大家的學(xué)習或工作具有一定的參考借鑒價值,需要的朋友可以參考下
1、發(fā)布確認高級
1. 存在的問題
再生產(chǎn)環(huán)境中由于一些不明原因?qū)е?code>rabbitmq重啟，在RabbitMQ重啟期間生產(chǎn)者消息投遞失敗，會導(dǎo)致消息丟失。
1.1、發(fā)布確認SpringBoot版本
1.1.1、確認機制方案
當消息不能正常被接收的時候，我們需要將消息存放在緩存中。
1.1.2、代碼架構(gòu)圖
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ā)回調(diào)方方法。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、回調(diào)接口
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;
/**
* 回調(diào)接口
*/
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
@Autowired
RabbitTemplate rabbitTemplate;
@PostConstruct
public void init(){
rabbitTemplate.setConfirmCallback(this);
}
/**
* 交換機接受失敗后進行回調(diào)
* 1. 保存消息的ID及相關(guān)消息
* 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("交換機已經(jīng)收到id為：{}的消息",id);
}else{
log.info("交換機還未收到id為：{}消息，由于原因：{}",id,s);
}
}
}
1.1.6、生產(chǎn)者
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ā)送得內(nèi)容是：{}",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、測試結(jié)果
1. 第一種情況
ID為1的消息正常送達，ID為2的消息由于RoutingKey的錯誤，導(dǎo)致不能正常被消費，但是交換機還是正常收到了消息，所以此時由于交換機正常接收之后的原因丟失的消息不能正常被接收。
2. 第二種情況
我們再上一種情況下修改了ID為1的消息的交換機的名稱，所以此時回調(diào)函數(shù)會進行回答由于什么原因?qū)е陆粨Q機無法接收成功消息。
1.2、回退消息
1.2.1、Mandatory參數(shù)
- 在僅開啟了生產(chǎn)者確認機制的情況下，交換機接收到消息后，會直接給消息生產(chǎn)者發(fā)送確認消息，如果發(fā)現(xiàn)該消息不可路由（就是消息被交換機成功接收后，無法到達隊列），那么消息會直接被丟棄，此時生產(chǎn)者是不知道消息被丟棄這個事件的。
- 通過設(shè)置該參數(shù)可以在消息傳遞過程中不可達目的地時將消息返回給生產(chǎn)者。
1.2.2、配置文件
spring.rabbitmq.publisher-returns=true需要在配置文件種開啟返回回調(diào)
1.2.3、生產(chǎn)者代碼
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ā)送得內(nèi)容是：{}",message + routingKey1);
CorrelationData correlationData2 = new CorrelationData("2");
String routingKey2 = "key2";
rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey2,message + routingKey2,correlationData2);
log.info("發(fā)送得內(nèi)容是：{}",message + routingKey2);
}
}
1.2.4、回調(diào)接口代碼
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;
/**
* 回調(diào)接口
*/
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
@Autowired
RabbitTemplate rabbitTemplate;
@PostConstruct
public void init(){
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
/**
* 交換機接受失敗后進行回調(diào)
* 1. 保存消息的ID及相關(guān)消息
* 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("交換機已經(jīng)收到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é)果
其他類的代碼與上一小節(jié)案例相同
ID為2的消息由于RoutingKey不可路由，但是還是被回調(diào)函數(shù)處理了。
1.3、備份交換機
1.3.1、代碼架構(gòu)圖
這里我們新增了備份交換機、備份隊列、報警隊列。它們綁定關(guān)系如圖所示。如果確認交換機成功接收的消息無法路由到相應(yīng)的隊列，就會被確認交換機發(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ā)現(xiàn)不可路由的消息內(nèi)容為：{}",msg);
}
}
1.3.4、測試結(jié)果
mandatory參數(shù)與備份交換機可以一起使用的時候，如果兩者同時開啟，備份交換機優(yōu)先級高。
到此這篇關(guān)于RabbitMQ發(fā)布確認高級的文章就介紹到這了,更多相關(guān)RabbitMQ發(fā)布確認高級內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
yapi是一個很好的接口文檔維護工具,其swagger功能,可將接口信息同步到y(tǒng)api平臺上,這篇文章主要介紹了idea同步y(tǒng)api插件,需要的朋友可以參考下2024-04-04
Javas使用Redlock實現(xiàn)分布式鎖過程解析
這篇文章主要介紹了Javas使用Redlock實現(xiàn)分布式鎖過程解析,文中通過示例代碼介紹的非常詳細，對大家的學(xué)習或者工作具有一定的參考學(xué)習價值,需要的朋友可以參考下2019-08-08
@Value注解是Spring框架中強大且常用的注解之一,本文主要介紹了SpringBoot中@value的常見用法及案例,具有一定的參考價值,感興趣的可以了解一下2023-09-09
這篇文章主要介紹了詳解Java編程中throw和throws子句的使用方法,是Java入門學(xué)習中的基礎(chǔ)知識,需要的朋友可以參考下2015-09-09
java實現(xiàn)字符串匹配求兩個字符串的最大公共子串
這篇文章主要介紹了java實現(xiàn)求兩個字符串最大公共子串的方法，詳細的描述了兩個字符串的最大公共子串算法的實現(xiàn),需要的朋友可以參考下2016-10-10
java同步器AQS架構(gòu)AbstractQueuedSynchronizer原理解析下
這篇文章主要為大家介紹了java同步器AQS架構(gòu)AbstractQueuedSynchronizer原理解析下，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步2022-03-03
SpringBoot全局異常與數(shù)據(jù)校驗的方法
這篇文章主要介紹了SpringBoot全局異常與數(shù)據(jù)校驗的方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-11-11
之前做這個頭像上傳功能還是花了好多時間的,今天我將我的代碼分享給大家,下面這篇文章主要給大家介紹了關(guān)于Java編程之如何通過JSP實現(xiàn)頭像自定義上傳的相關(guān)資料,文中通過示例代碼介紹的非常詳細,需要的朋友可以參考下2022-12-12
Intellij?IDEA創(chuàng)建web項目的超詳細步驟記錄
如果剛開始接觸IDEA,或者之前使用的是eclipse/myEclipse的話,即使是創(chuàng)建一個JAVA WEB項目,估計也讓很多人費了好幾個小時,下面這篇文章主要給大家介紹了關(guān)于Intellij?IDEA創(chuàng)建web項目的超詳細步驟,需要的朋友可以參考下2022-08-08
SpringBoot+Vue實現(xiàn)EasyPOI導(dǎo)入導(dǎo)出的方法詳解
項目開發(fā)過程中，很大的需求都有 導(dǎo)入導(dǎo)出功能。本文將利用SpringBoot+Vue實現(xiàn)EasyPOI導(dǎo)入導(dǎo)出功能，感興趣的可以了解一下2022-08-08