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