SpringBoot整合RabbitMQ消息隊列的完整步驟
更新時間：2021年05月06日 11:03:52 作者：JackHorse
這篇文章主要給大家介紹了關于SpringBoot整合RabbitMQ消息隊列的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
SpringBoot整合RabbitMQ
主要實現(xiàn)RabbitMQ以下三種消息隊列：
- 簡單消息隊列（演示direct模式）
- 基于RabbitMQ特性的延時消息隊列
- 基于RabbitMQ相關插件的延時消息隊列
公共資源
1. 引入pom依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2. 配置yml文件
基于上篇《RabbitMQ安裝與配置》實現(xiàn)的情況下，進行基礎配置。
spring:
rabbitmq:
host: 121.5.168.31
port: 5672 # 默認可省略
virtual-host: /*** # 虛擬主機
username: *** # 用戶名
password: *** # 用戶密碼
# 開啟投遞成功回調 P -> Exchange
publisher-confirm-type: correlated
# 開啟投遞消息到隊列失敗回調 Exchange -> Queue
publisher-returns: true
# 開啟手動ACK確認模式 Queue -> C
listener:
simple:
acknowledge-mode: manual # 代表手動ACK確認
# 一些基本參數(shù)的設置
concurrency: 3
prefetch: 15
retry:
enabled: true
max-attempts: 5
max-concurrency: 10
3. 公共Constants類
/**
* @author Mr.Horse
* @version 1.0
* @description: {description}
* @date 2021/4/23 15:28
*/
public class Constants {
/**
* 第一個配置Queue,Exchange,Key(非注解方式)
*/
public final static String HORSE_SIMPLE_QUEUE = "HORSE_SIMPLE_QUEUE";
public final static String HORSE_SIMPLE_EXCHANGE = "HORSE_SIMPLE_EXCHANGE";
public final static String HORSE_SIMPLE_KEY = "HORSE_SIMPLE_KEY";
/**
* 第二個配置Queue,Exchange,Key（注解方式）
*/
public final static String HORSE_ANNOTATION_QUEUE = "HORSE_ANNOTATION_QUEUE";
public final static String HORSE_ANNOTATION_EXCHANGE = "HORSE_ANNOTATION_EXCHANGE";
public final static String HORSE_ANNOTATION_KEY = "HORSE_ANNOTATION_KEY";
//************************************延時消息隊列配置信息**************************
/**
* 延時隊列信息配置
*/
public final static String HORSE_DELAY_EXCHANGE = "HORSE_DELAY_EXCHANGE";
public final static String HORSE_DELAY_QUEUE = "HORSE_DELAY_QUEUE";
public final static String HORSE_DELAY_KEY = "HORSE_DELAY_KEY";
/**
* 死信隊列
*/
public final static String HORSE_DEAD_EXCHANGE = "HORSE_DEAD_EXCHANGE";
public final static String HORSE_DEAD_QUEUE = "HORSE_DEAD_QUEUE";
public final static String HORSE_DEAD_KEY = "HORSE_DEAD_KEY";
//**************************************延時消息隊列配置信息(插件版)******************************
/**
* 新延時隊列信息配置
*/
public final static String HORSE_PLUGIN_EXCHANGE = "HORSE_PLUGIN_EXCHANGE";
public final static String HORSE_PLUGIN_QUEUE = "HORSE_PLUGIN_QUEUE";
public final static String HORSE_PLUGIN_KEY = "HORSE_PLUGIN_KEY";
}
簡單消息隊列（direct模式）
4. RabbitTemplate模板配置
主要定義消息投遞Exchange成功回調函數(shù)和消息從Exchange投遞到消息隊列失敗的回調函數(shù)。
package com.topsun.rabbit;
import com.sun.org.apache.xpath.internal.operations.Bool;
import com.topsun.constants.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Mr.Horse
* @version 1.0
* @description: {description}
* @date 2021/4/23 14:17
*/
@Configuration
public class RabbitConfig {
private static Logger logger = LoggerFactory.getLogger(RabbitConfig.class);
@Autowired
private CachingConnectionFactory connectionFactory;
/**
* @return
*/
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// 觸發(fā)setReturnCallback回調必須設置mandatory=true,否則Exchange沒有找到Queue就會丟棄掉消息, 而不會觸發(fā)回調
rabbitTemplate.setMandatory(Boolean.TRUE);
// 設置序列化機制
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
// 消息由投遞到Exchange中時觸發(fā)的回調
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
logger.info("消息發(fā)送到Exchange情況反饋:唯一標識：correlationData={},消息確認：ack={},原因：cause={}",
correlationData, ack, cause)
);
// 消息由Exchange發(fā)送到Queue時失敗觸發(fā)的回調
rabbitTemplate.setReturnsCallback((returnedMessage) -> {
// 如果是插件形式實現(xiàn)的延時隊列,則直接返回
// 原因: 因為發(fā)送方確實沒有投遞到隊列上，只是在交換器上暫存，等過期時間到了 才會發(fā)往隊列,從而實現(xiàn)延時隊列的操作
if (Constants.HORSE_PLUGIN_EXCHANGE.equals(returnedMessage.getExchange())) {
return;
}
logger.warn("消息由Exchange發(fā)送到Queue時失敗:message={},replyCode={},replyText={},exchange={},rountingKey={}",
returnedMessage.getMessage(), returnedMessage.getReplyText(), returnedMessage.getReplyText(),
returnedMessage.getExchange(), returnedMessage.getRoutingKey());
});
return rabbitTemplate;
}
//*******************************************直接配置綁定關系*****************************************
/**
* 聲明隊列
*
* @return
*/
@Bean
public Queue horseQueue() {
return new Queue(Constants.HORSE_SIMPLE_QUEUE, Boolean.TRUE);
}
/**
* 聲明指定模式交換機
*
* @return
*/
@Bean
public DirectExchange horseExchange() {
return new DirectExchange(Constants.HORSE_SIMPLE_EXCHANGE, Boolean.TRUE, Boolean.FALSE);
}
/**
* 綁定交換機,隊列,路由Key
*
* @return
*/
@Bean
public Binding horseBinding() {
return BindingBuilder.bind(horseQueue()).to(horseExchange()).with(Constants.HORSE_SIMPLE_KEY);
}
}
5. 定義消息監(jiān)聽器
基于 @RabbitListenerzi注解，實現(xiàn)自定義消息監(jiān)聽器。主要有兩種實現(xiàn)方式：
- 如果在配置類中聲明了Queue、Excehange以及他們直接的綁定，這里直接指定隊列進行消息監(jiān)聽
- 如果前面什么也沒做，這里可以直接用注解的方式進行綁定實現(xiàn)消息監(jiān)聽
package com.topsun.rabbit;
import com.rabbitmq.client.Channel;
import com.topsun.constants.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
* @author Mr.Horse
* @version 1.0
* @description: {description}
* @date 2021/4/23 14:58
*/
@Component
public class MsgListener {
private static Logger logger = LoggerFactory.getLogger(MsgListener.class);
/**
* 配置類中已經完成綁定,這里直接根據(jù)隊列值接收
*
* @param message
* @param channel
* @param msg
*/
@RabbitListenerzi(queues = Constants.HORSE_SIMPLE_QUEUE)
public void customListener(Message message, Channel channel, String msg) {
// 獲取每條消息唯一標識(用于手動ACK確認)
long tag = message.getMessageProperties().getDeliveryTag();
try {
logger.info(" ==> customListener接收" + msg);
// 手動ACK確認
channel.basicAck(tag, false);
} catch (IOException e) {
logger.error(" ==> 消息接收失敗: {}", tag);
}
}
/**
* 根據(jù)注解的形式進行綁定接收
*
* @param message
* @param channel
* @param msg
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = Constants.HORSE_ANNOTATION_QUEUE, durable = "true"),
exchange = @Exchange(value = Constants.HORSE_ANNOTATION_EXCHANGE, ignoreDeclarationExceptions = "true"),
key = {Constants.HORSE_ANNOTATION_KEY}
))
public void annotationListener(Message message, Channel channel, String msg) {
// 獲取每條消息唯一標識(用于手動ACK確認)
long tag = message.getMessageProperties().getDeliveryTag();
try {
logger.info(" ==> annotationListener接收" + msg);
// 手動ACK確認
channel.basicAck(tag, false);
} catch (IOException e) {
logger.error(" ==> 消息接收失敗: {}", tag);
}
}
}
6. 測試接口
這里發(fā)送100條消息：
- 奇數(shù)條到非注解方式的消息監(jiān)聽器
- 偶數(shù)條到注解式消息監(jiān)聽器
@GetMapping("/rabbit")
public void sendMsg() {
for (int i = 1; i <= 100; i++) {
String msg = "第" + i + "條消息";
logger.info("==> 發(fā)送" + msg);
if (i % 2 == 1) {
rabbitTemplate.convertAndSend(Constants.HORSE_SIMPLE_EXCHANGE, Constants.HORSE_SIMPLE_KEY, msg, new CorrelationData(String.valueOf(i)));
} else {
rabbitTemplate.convertAndSend(Constants.HORSE_ANNOTATION_EXCHANGE, Constants.HORSE_ANNOTATION_KEY, msg, new CorrelationData(String.valueOf(i)));
}
}
}
結果：自行測試過，非常成功:smile::smile::smile:
延時消息隊列
原理：生產者生產一條延時消息，根據(jù)需要延時時間的不同，利用不同的routingkey將消息路由到不同的延時隊列，每個隊列都設置了不同的TTL屬性，并綁定在同一個死信交換機中，消息過期后，根據(jù)routingkey的不同，又會被路由到不同的死信隊列中，消費者只需要監(jiān)聽對應的死信隊列進行處理即可。
7. 配置綁定相關信息
/**
* @author Mr.Horse
* @version 1.0
* @description: {description}
* @date 2021/4/24 14:22
*/
@Configuration
public class DelayRabbitConfig {
private static Logger logger = LoggerFactory.getLogger(DelayRabbitConfig.class);
/**
* 聲明延時隊列交換機
*
* @return
*/
@Bean
public DirectExchange delayExchange() {
return new DirectExchange(Constants.HORSE_DELAY_EXCHANGE, Boolean.TRUE, Boolean.FALSE);
}
/**
* 聲明死信隊列交換機
*
* @return
*/
@Bean
public DirectExchange deadExchange() {
return new DirectExchange(Constants.HORSE_DEAD_EXCHANGE, Boolean.TRUE, Boolean.FALSE);
}
/**
* 聲明延時隊列 延時10s(單位:ms),并將延時隊列綁定到對應的死信交換機和路由Key
*
* @return
*/
@Bean
public Queue delayQueue() {
Map<String, Object> args = new HashMap<>(3);
// x-dead-letter-exchange 這里聲明當前隊列綁定的死信交換機
args.put("x-dead-letter-exchange", Constants.HORSE_DEAD_EXCHANGE);
// x-dead-letter-routing-key 這里聲明當前隊列的死信路由key
args.put("x-dead-letter-routing-key", Constants.HORSE_DEAD_KEY);
// x-message-ttl 聲明隊列的TTL(過期時間)
// 可以在這里直接寫死，也可以進行動態(tài)的設置（推薦動態(tài)設置）
// args.put("x-message-ttl", 10000);
return QueueBuilder.durable(Constants.HORSE_DELAY_QUEUE).withArguments(args).build();
}
/**
* 聲明死信隊列
*
* @return
*/
@Bean
public Queue deadQueue() {
return new Queue(Constants.HORSE_DEAD_QUEUE, Boolean.TRUE);
}
/**
* 延時隊列綁定管理
*
* @return
*/
@Bean
public Binding delayBinding() {
return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(Constants.HORSE_DELAY_KEY);
}
/**
* 死信隊列綁定管理
*
* @return
*/
@Bean
public Binding deadBinding() {
return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(Constants.HORSE_DEAD_KEY);
}
//**********************************延時消息隊列配置信息(插件版)************************************
@Bean
public Queue pluginQueue() {
return new Queue(Constants.HORSE_PLUGIN_QUEUE);
}
/**
* 設置延時隊列的交換機，必須是 CustomExchange 類型交換機
* 參數(shù)必須，不能改變
* @return
*/
@Bean
public CustomExchange customPluginExchange() {
Map<String, Object> args = new HashMap<>(2);
args.put("x-delayed-type", "direct");
return new CustomExchange(Constants.HORSE_PLUGIN_EXCHANGE, "x-delayed-message", Boolean.TRUE, Boolean.FALSE, args);
}
@Bean
public Binding pluginBinding() {
return BindingBuilder.bind(pluginQueue()).to(customPluginExchange()).with(Constants.HORSE_PLUGIN_KEY).noargs();
}
}
8. 定義延時監(jiān)聽器
/**
* @author Mr.Horse
* @version 1.0
* @description: {description}
* @date 2021/4/24 14:51
*/
@Component
public class DelayMsgListener {
private static Logger logger = LoggerFactory.getLogger(DelayMsgListener.class);
/**
* 監(jiān)聽死信隊列
*
* @param message
* @param channel
* @param msg
*/
@RabbitListener(queues = Constants.HORSE_DEAD_QUEUE)
public void consumeDeadListener(Message message, Channel channel, String msg) {
long tag = message.getMessageProperties().getDeliveryTag();
try {
logger.info(" ==> consumeDeadListener接收" + msg);
// 手動ACK確認
channel.basicAck(tag, false);
} catch (IOException e) {
logger.error(" ==> 消息接收失敗: {}", tag);
}
}
/**
* 監(jiān)聽延時隊列(插件版)
*
* @param message
* @param channel
* @param msg
*/
@RabbitListener(queues = Constants.HORSE_PLUGIN_QUEUE)
public void consumePluginListener(Message message, Channel channel, String msg) {
long tag = message.getMessageProperties().getDeliveryTag();
try {
logger.info(" ==> consumePluginListener" + msg);
// 手動ACK確認
channel.basicAck(tag, false);
} catch (IOException e) {
logger.error(" ==> 消息接收失敗: {}", tag);
}
}
}
9. 測試接口
// 基于特性的延時隊列
@GetMapping("/delay/rabbit")
public void delayMsg(@RequestParam("expire") Long expire) {
for (int i = 1; i <= 10; i++) {
String msg = "第" + i + "條消息";
logger.info("==> 發(fā)送" + msg);
// 這里可以動態(tài)的設置過期時間
rabbitTemplate.convertAndSend(Constants.HORSE_DELAY_EXCHANGE, Constants.HORSE_DELAY_KEY, msg,
message -> {
message.getMessageProperties().setExpiration(String.valueOf(expire));
return message;
},
new CorrelationData(String.valueOf(i)));
}
}
// 基于插件的延時隊列
@GetMapping("/delay/plugin")
public void delayPluginMsg(@RequestParam("expire") Integer expire) {
for (int i = 1; i <= 10; i++) {
String msg = "第" + i + "條消息";
logger.info("==> 發(fā)送" + msg);
// 動態(tài)設置過期時間
rabbitTemplate.convertAndSend(Constants.HORSE_PLUGIN_EXCHANGE, Constants.HORSE_PLUGIN_KEY, msg, message -> {
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
message.getMessageProperties().setDelay(expire);
return message;
}, new CorrelationData(String.valueOf(i)));
}
}
結果：你懂的:scream_cat::scream_cat::scream_cat:
RabbitMQ的基礎使用演示到此結束。
總結
到此這篇關于SpringBoot整合RabbitMQ消息隊列的文章就介紹到這了,更多相關SpringBoot整合RabbitMQ消息隊列內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
研究表明，Java堆中對象占據(jù)最大比重的就是字符串對象，所以弄清楚字符串知識對學習Java很重要。本文主要重點聊聊字符串常量池，希望對大家有所幫助2022-12-12
JSON.toJSONString()方法在Java中的使用方法及應用場景
這篇文章主要給大家介紹了關于JSON.toJSONString()方法在Java中的使用方法及應用場景,JSON.toJSONString是將對象轉化為Json字符串,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2024-04-04
Java調用wsdl接口的兩種方法(axis和wsimport)
本文主要介紹了Java調用wsdl接口的兩種方法(axis和wsimport)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-03-03
這篇文章主要介紹了純注解版spring與mybatis的整合過程，本文通過示例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2022-06-06
Retrofit+RxJava實現(xiàn)帶進度下載文件
這篇文章主要為大家詳細介紹了Retrofit+RxJava實現(xiàn)帶進度下載文件，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-05-05
SpringBoot與Quartz集成實現(xiàn)分布式定時任務集群的代碼實例
今天小編就為大家分享一篇關于SpringBoot與Quartz集成實現(xiàn)分布式定時任務集群的代碼實例，小編覺得內容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2019-03-03
解析阿里一面CyclicBarrier和CountDownLatch的區(qū)別
這篇文章主要介紹了阿里一面CyclicBarrier和CountDownLatch的區(qū)別是啥,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-03-03
這篇文章主要介紹了剖析Java中阻塞隊列的實現(xiàn)原理及應用場景,這里也對阻塞和非阻塞隊列的不同之處進行了對比,需要的朋友可以參考下2015-12-12
這篇文章主要介紹了Messges Queue消息隊列詳解,消息隊列一般簡稱為 MQ,是指利用高效可靠的消息傳遞機制進行與平臺無關的數(shù)據(jù)交流,并基于數(shù)據(jù)通信來進行分布式系統(tǒng)的集成,是在消息的傳輸過程中保存消息的容器,需要的朋友可以參考下2023-07-07
這篇文章主要介紹了2020版IDEA整合GitHub的方法，本文通過圖文并茂的形式給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2020-08-08