SpringBoot整合RabbitMQ實(shí)現(xiàn)流量消峰
引言
RabbitMQ 即一個(gè)消息隊(duì)列，主要是用來實(shí)現(xiàn)應(yīng)用程序的異步和解耦，同時(shí)也能起到消息緩沖，消息分發(fā)的作用。
消息中間件在互聯(lián)網(wǎng)公司的使用中越來越多，剛才還看到新聞阿里將 RocketMQ 捐獻(xiàn)給了 Apache，當(dāng)然了今天的主角還是講 RabbitMQ。消息中間件最主要的作用是解耦，中間件最標(biāo)準(zhǔn)的用法是生產(chǎn)者生產(chǎn)消息傳送到隊(duì)列，消費(fèi)者從隊(duì)列中拿取消息并處理，生產(chǎn)者不用關(guān)心是誰來消費(fèi)，消費(fèi)者不用關(guān)心誰在生產(chǎn)消息，從而達(dá)到解耦的目的。在分布式的系統(tǒng)中，消息隊(duì)列也會(huì)被用在很多其它的方面，比如：分布式事務(wù)的支持，RPC 的調(diào)用等等。
以前一直使用的是 ActiveMQ，在實(shí)際的生產(chǎn)使用中也出現(xiàn)了一些小問題，在網(wǎng)絡(luò)查閱了很多的資料后，決定嘗試使用 RabbitMQ 來替換 ActiveMQ，RabbitMQ 的高可用性、高性能、靈活性等一些特點(diǎn)吸引了我們，查閱了一些資料整理出此文。
RabbitMQ 介紹
RabbitMQ 是實(shí)現(xiàn) AMQP（高級消息隊(duì)列協(xié)議）的消息中間件的一種，最初起源于金融系統(tǒng)，用于在分布式系統(tǒng)中存儲轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。RabbitMQ 主要是為了實(shí)現(xiàn)系統(tǒng)之間的雙向解耦而實(shí)現(xiàn)的。當(dāng)生產(chǎn)者大量產(chǎn)生數(shù)據(jù)時(shí)，消費(fèi)者無法快速消費(fèi)，那么需要一個(gè)中間層。保存這個(gè)數(shù)據(jù)。
AMQP，即 Advanced Message Queuing Protocol，高級消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、隊(duì)列、路由（包括點(diǎn)對點(diǎn)和發(fā)布/訂閱）、可靠性、安全。
RabbitMQ 是一個(gè)開源的 AMQP 實(shí)現(xiàn)，服務(wù)器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系統(tǒng)中存儲轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。
相關(guān)概念
通常我們談到隊(duì)列服務(wù), 會(huì)有三個(gè)概念：發(fā)消息者、隊(duì)列、收消息者，RabbitMQ 在這個(gè)基本概念之上, 多做了一層抽象, 在發(fā)消息者和 隊(duì)列之間, 加入了交換器 (Exchange). 這樣發(fā)消息者和隊(duì)列就沒有直接聯(lián)系, 轉(zhuǎn)而變成發(fā)消息者把消息給交換器, 交換器根據(jù)調(diào)度策略再把消息再給隊(duì)列。
左側(cè) P 代表 生產(chǎn)者，也就是往 RabbitMQ 發(fā)消息的程序。
中間即是 RabbitMQ，其中包括了 交換機(jī) 和 隊(duì)列。
右側(cè) C 代表 消費(fèi)者，也就是往 RabbitMQ 拿消息的程序。
那么，其中比較重要的概念有 4 個(gè)，分別為：虛擬主機(jī)，交換機(jī)，隊(duì)列，和綁定。
虛擬主機(jī)：一個(gè)虛擬主機(jī)持有一組交換機(jī)、隊(duì)列和綁定。為什么需要多個(gè)虛擬主機(jī)呢？很簡單， RabbitMQ 當(dāng)中，用戶只能在虛擬主機(jī)的粒度進(jìn)行權(quán)限控制。因此，如果需要禁止A組訪問B組的交換機(jī)/隊(duì)列/綁定，必須為A和B分別創(chuàng)建一個(gè)虛擬主機(jī)。每一個(gè) RabbitMQ 服務(wù)器都有一個(gè)默認(rèn)的虛擬主機(jī)“/”。
交換機(jī)：Exchange 用于轉(zhuǎn)發(fā)消息，但是它不會(huì)做存儲 ，如果沒有 Queue bind 到 Exchange 的話，它會(huì)直接丟棄掉 Producer 發(fā)送過來的消息。這里有一個(gè)比較重要的概念：路由鍵 。消息到交換機(jī)的時(shí)候，交互機(jī)會(huì)轉(zhuǎn)發(fā)到對應(yīng)的隊(duì)列中，那么究竟轉(zhuǎn)發(fā)到哪個(gè)隊(duì)列，就要根據(jù)該路由鍵。
綁定：也就是交換機(jī)需要和隊(duì)列相綁定，這其中如上圖所示，是多對多的關(guān)系。
交換機(jī)(Exchange)
交換機(jī)的功能主要是接收消息并且轉(zhuǎn)發(fā)到綁定的隊(duì)列，交換機(jī)不存儲消息，在啟用ack模式后，交換機(jī)找不到隊(duì)列會(huì)返回錯(cuò)誤。交換機(jī)有四種類型：Direct, topic, Headers and Fanout
Direct：direct 類型的行為是"先匹配, 再投送". 即在綁定時(shí)設(shè)定一個(gè) routingkey, 消息的routingkey 匹配時(shí), 才會(huì)被交換器投送到綁定的隊(duì)列中去.
Topic：按規(guī)則轉(zhuǎn)發(fā)消息（最靈活）
Headers：設(shè)置 header attribute 參數(shù)類型的交換機(jī)
Fanout：轉(zhuǎn)發(fā)消息到所有綁定隊(duì)列
Direct Exchange
Direct Exchange 是 RabbitMQ 默認(rèn)的交換機(jī)模式，也是最簡單的模式，根據(jù)key全文匹配去尋找隊(duì)列。
第一個(gè) X - Q1 就有一個(gè) binding key，名字為 orange；X - Q2 就有 2 個(gè) binding key，名字為 black 和 green。當(dāng)消息中的 路由鍵 和 這個(gè) binding key 對應(yīng)上的時(shí)候，那么就知道了該消息去到哪一個(gè)隊(duì)列中。
Ps：為什么 X 到 Q2 要有 black，green，2個(gè) binding key呢，一個(gè)不就行了嗎？- 這個(gè)主要是因?yàn)榭赡苡钟?Q3，而Q3只接受 black 的信息，而Q2不僅接受black 的信息，還接受 green 的信息。
Topic Exchange
Topic Exchange 轉(zhuǎn)發(fā)消息主要是根據(jù)通配符。在這種交換機(jī)下，隊(duì)列和交換機(jī)的綁定會(huì)定義一種路由模式，那么，通配符就要在這種路由模式和路由鍵之間匹配后交換機(jī)才能轉(zhuǎn)發(fā)消息。
在這種交換機(jī)模式下：
路由鍵必須是一串字符，用句號（ .） 隔開，比如說 agreements.us，或者 agreements.eu.stockholm 等。
路由模式必須包含一個(gè) 星號（
*），主要用于匹配路由鍵指定位置的一個(gè)單詞，比如說，一個(gè)路由模式是這樣子：agreements..b.*，那么就只能匹配路由鍵是這樣子的：第一個(gè)單詞是 agreements，第四個(gè)單詞是 b。井號（#）就表示相當(dāng)于一個(gè)或者多個(gè)單詞，例如一個(gè)匹配模式是 agreements.eu.berlin.#，那么，以agreements.eu.berlin 開頭的路由鍵都是可以的。
具體代碼發(fā)送的時(shí)候還是一樣，第一個(gè)參數(shù)表示交換機(jī)，第二個(gè)參數(shù)表示 routing key，第三個(gè)參數(shù)即消息。如下：
rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is RabbitMQ!");
topic 和 direct 類似, 只是匹配上支持了"模式", 在"點(diǎn)分"的 routing_key 形式中, 可以使用兩個(gè)通配符:
*表示一個(gè)詞.#表示零個(gè)或多個(gè)詞.
Headers Exchange
headers 也是根據(jù)規(guī)則匹配, 相較于 direct 和 topic 固定地使用 routing_key , headers 則是一個(gè)自定義匹配規(guī)則的類型. 在隊(duì)列與交換器綁定時(shí), 會(huì)設(shè)定一組鍵值對規(guī)則, 消息中也包括一組鍵值對( headers 屬性), 當(dāng)這些鍵值對有一對, 或全部匹配時(shí), 消息被投送到對應(yīng)隊(duì)列.
Fanout Exchange
Fanout Exchange 消息廣播的模式，不管路由鍵或者是路由模式，會(huì)把消息發(fā)給綁定給它的全部隊(duì)列，如果配置了 routing_key 會(huì)被忽略。
Spring Boot 集成 RabbitMQ
Spring Boot 集成 RabbitMQ 非常簡單，如果只是簡單的使用配置非常少，Spring Boot 提供了 spring-boot-starter-amqp 項(xiàng)目對消息各種支持。
簡單使用
1、配置 Pom 包，主要是添加 spring-boot-starter-amqp 的支持
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>2、配置文件
配置 RabbitMQ 的安裝地址、端口以及賬戶信息
spring.application.name=Spring-boot-rabbitmq spring.rabbitmq.host=192.168.0.86 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=123456
3、隊(duì)列配置
@Configuration
public class RabbitConfig {
@Bean
public Queue Queue() {
return new Queue("hello");
}
}4、發(fā)送者
rabbitTemplate 是 Spring Boot 提供的默認(rèn)實(shí)現(xiàn)
@component
public class HelloSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String context = "hello " + new Date();
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("hello", context);
}
}4、接收者
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver : " + hello);
}
}5、測試
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {
@Autowired
private HelloSender helloSender;
@Test
public void hello() throws Exception {
helloSender.send();
}
}注意，發(fā)送者和接收者的 queue name 必須一致，不然不能接收
多對多使用
一個(gè)發(fā)送者，N 個(gè)接收者或者 N 個(gè)發(fā)送者和 N 個(gè)接收者會(huì)出現(xiàn)什么情況呢？
一對多發(fā)送
對上面的代碼進(jìn)行了小改造，接收端注冊了兩個(gè) Receiver,Receiver1 和 Receiver2，發(fā)送端加入?yún)?shù)計(jì)數(shù)，接收端打印接收到的參數(shù)，下面是測試代碼，發(fā)送一百條消息，來觀察兩個(gè)接收端的執(zhí)行效果
@Test
public void oneToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
}
}結(jié)果如下：
Receiver 1: Spring boot test queue ****** 11 Receiver 2: Spring boot test queue ****** 12 Receiver 2: Spring boot test queue ****** 14 Receiver 1: Spring boot test queue ****** 13 Receiver 2: Spring boot test queue ****** 15 Receiver 1: Spring boot test queue ****** 16 Receiver 1: Spring boot test queue ****** 18 Receiver 2: Spring boot test queue ****** 17 Receiver 2: Spring boot test queue ****** 19 Receiver 1: Spring boot test queue ****** 20
根據(jù)返回結(jié)果得到以下結(jié)論
一個(gè)發(fā)送者，N個(gè)接受者,經(jīng)過測試會(huì)均勻的將消息發(fā)送到N個(gè)接收者中
多對多發(fā)送
復(fù)制了一份發(fā)送者，加入標(biāo)記，在一百個(gè)循環(huán)中相互交替發(fā)送
@Test
public void manyToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
neoSender2.send(i);
}
}結(jié)果如下：
Receiver 1: Spring boot test queue ****** 20 Receiver 2: Spring boot test queue ****** 20 Receiver 1: Spring boot test queue ****** 21 Receiver 2: Spring boot test queue ****** 21 Receiver 1: Spring boot test queue ****** 22 Receiver 2: Spring boot test queue ****** 22 Receiver 1: Spring boot test queue ****** 23 Receiver 2: Spring boot test queue ****** 23 Receiver 1: Spring boot test queue ****** 24 Receiver 2: Spring boot test queue ****** 24 Receiver 1: Spring boot test queue ****** 25 Receiver 2: Spring boot test queue ****** 25
結(jié)論：和一對多一樣，接收端仍然會(huì)均勻接收到消息
高級使用
對象的支持
Spring Boot 以及完美的支持對象的發(fā)送和接收，不需要格外的配置。
//發(fā)送者
public void send(User user) {
System.out.println("Sender object: " + user.toString());
this.rabbitTemplate.convertAndSend("object", user);
}
...
//接收者
@RabbitHandler
public void process(User user) {
System.out.println("Receiver object : " + user);
}結(jié)果如下：
Sender object: User{name='test', pass='123456'}
Receiver object : User{name='test', pass='123456'}Topic Exchange
topic 是 RabbitMQ 中最靈活的一種方式，可以根據(jù) routing_key 自由的綁定不同的隊(duì)列
首先對 topic 規(guī)則配置，這里使用兩個(gè)隊(duì)列來測試
@Configuration
public class TopicRabbitConfig {
final static String message = "topic.message";
final static String messages = "topic.messages";
@Bean
public Queue queueMessage() {
return new Queue(TopicRabbitConfig.message);
}
@Bean
public Queue queueMessages() {
return new Queue(TopicRabbitConfig.messages);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("exchange");
}
@Bean
Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}
@Bean
Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}
}使用 queueMessages 同時(shí)匹配兩個(gè)隊(duì)列，queueMessage 只匹配 "topic.message" 隊(duì)列
public void send1() {
String context = "hi, i am message 1";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}
public void send2() {
String context = "hi, i am messages 2";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
}發(fā)送send1會(huì)匹配到topic.#和topic.message 兩個(gè)Receiver都可以收到消息，發(fā)送send2只有topic.#可以匹配所有只有Receiver2監(jiān)聽到消息
Fanout Exchange
Fanout 就是我們熟悉的廣播模式或者訂閱模式，給 Fanout 交換機(jī)發(fā)送消息，綁定了這個(gè)交換機(jī)的所有隊(duì)列都收到這個(gè)消息。
Fanout 相關(guān)配置
@Configuration
public class FanoutRabbitConfig {
@Bean
public Queue AMessage() {
return new Queue("fanout.A");
}
@Bean
public Queue BMessage() {
return new Queue("fanout.B");
}
@Bean
public Queue CMessage() {
return new Queue("fanout.C");
}
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
@Bean
Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
return BindingBuilder.bind(AMessage).to(fanoutExchange);
}
@Bean
Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(BMessage).to(fanoutExchange);
}
@Bean
Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(CMessage).to(fanoutExchange);
}
}這里使用了 A、B、C 三個(gè)隊(duì)列綁定到 Fanout 交換機(jī)上面，發(fā)送端的 routing_key 寫任何字符都會(huì)被忽略：
public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}這里使用了 A、B、C 三個(gè)隊(duì)列綁定到 Fanout 交換機(jī)上面，發(fā)送端的 routing_key 寫任何字符都會(huì)被忽略：
public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}結(jié)果如下：
Sender : hi, fanout msg ... fanout Receiver B: hi, fanout msg fanout Receiver A : hi, fanout msg fanout Receiver C: hi, fanout msg
結(jié)果說明，綁定到 fanout 交換機(jī)上面的隊(duì)列都收到了消息。
以上就是SpringBoot整合RabbitMQ實(shí)現(xiàn)流量消峰的詳細(xì)內(nèi)容，更多關(guān)于SpringBoot RabbitMQ流量消峰的資料請關(guān)注腳本之家其它相關(guān)文章！
- Springboot使用Rabbitmq的延時(shí)隊(duì)列+死信隊(duì)列實(shí)現(xiàn)消息延期消費(fèi)
- Springboot使用RabbitMQ實(shí)現(xiàn)關(guān)閉超時(shí)訂單(示例詳解)
- SpringBoot整合Canal+RabbitMQ監(jiān)聽數(shù)據(jù)變更詳解
- SpringBoot集成和使用RabbitMQ方式
- springboot整合RabbitMQ中死信隊(duì)列的實(shí)現(xiàn)
- SpringBoot 整合 RabbitMQ 的使用方式(代碼示例)
- springboot整合rabbitmq實(shí)現(xiàn)訂單超時(shí)取消案例分析
相關(guān)文章
這篇文章給大家介紹SpringBoot通過tractId操作日志鏈路跟蹤,通過tractId,即可完成對從一個(gè)請求進(jìn)入系統(tǒng)到請求結(jié)束的日志追蹤,本文給大家介紹的非常詳細(xì),需要的朋友參考下吧2023-10-10
springboot如何通過session實(shí)現(xiàn)單點(diǎn)登入詳解
單點(diǎn)登錄(SSO)的定義是在多個(gè)應(yīng)用系統(tǒng)中,用戶只需要登錄一次就可以訪問所有相互信任的應(yīng)用系統(tǒng),下面這篇文章主要給大家介紹了關(guān)于springboot如何通過session實(shí)現(xiàn)單點(diǎn)登入的相關(guān)資料,需要的朋友可以參考下2021-12-12
Java8?lambda表達(dá)式的10個(gè)實(shí)例講解
這篇文章主要介紹了Java8?lambda表達(dá)式的10個(gè)實(shí)例，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-03-03
IDEA 開發(fā)配置SparkSQL及簡單使用案例代碼
這篇文章主要介紹了IDEA 開發(fā)配置SparkSQL及簡單使用案例代碼,本文通過代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-08-08
這篇文章主要介紹了JDBC用法,較為詳細(xì)的分析了基于JDBC進(jìn)行數(shù)據(jù)庫操作的技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2014-12-12
Spring AI與DeepSeek實(shí)戰(zhàn)一之快速打造智能對話應(yīng)用
本文詳細(xì)介紹了如何通過SpringAI框架集成DeepSeek大模型,實(shí)現(xiàn)普通對話和流式對話功能,步驟包括申請API-KEY、項(xiàng)目搭建、配置API-KEY、創(chuàng)建ChatClient對象、創(chuàng)建對話接口、切換模型、使用prompt模板、流式對話等,感興趣的朋友一起看看吧2025-03-03
Java 實(shí)現(xiàn)實(shí)時(shí)監(jiān)聽文件夾是否有新文件增加并上傳服務(wù)器功能
本文中主要陳述一種實(shí)時(shí)監(jiān)聽文件夾中是否有文件增加的功能，可用于實(shí)際文件上傳功能的開發(fā)。本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，具有一定的參考借鑒價(jià)值，需要的朋友參考下吧2019-09-09
Redis的字典緩存是支撐其高性能的一個(gè)關(guān)鍵因素,它使得Redis能夠以極快的速度處理大量的數(shù)據(jù),這篇文章主要介紹了springboot集成redis之字典緩存,需要的朋友可以參考下2024-07-07
Java ArrayList與LinkedList使用方法詳解
Java中容器對象主要用來存儲其他對象，根據(jù)實(shí)現(xiàn)原理不同，主要有3類常用的容器對象：ArrayList使用數(shù)組結(jié)構(gòu)存儲容器中的元素、LinkedList使用鏈表結(jié)構(gòu)存儲容器中的元素2022-11-11
SpringBoot支持多種請求方式,包括GET、RESTful、分頁列表等,常用框架如阿里fastjson、谷歌gson用于JavaBean序列化為Json,性能上Jackson最快,SpringBoot目錄結(jié)構(gòu)包括src/main/java、src/main/resources等2025-01-01