目錄
一、SpringAMQP
1.1、概念
1.2、前置知識(shí)（實(shí)現(xiàn)案例前必看?。?/p>
1.2.1、創(chuàng)建隊(duì)列
1.2.2、創(chuàng)建交換機(jī)
1.2.3、創(chuàng)建綁定
1.2.4、@RabbitListener 注解
a）情況一：queue 存在
b）情況二：queue 不存在?
1.2.5、為什么更建議使用 @Bean 注解創(chuàng)建，而不是 @RabbitListener 注解創(chuàng)建？
1.3、案例實(shí)現(xiàn)
1.3.1、SpringAMQP 實(shí)現(xiàn)基礎(chǔ)消息隊(duì)列（BasicQueue）
1.3.2、SpringAMQP 實(shí)現(xiàn)工作消息隊(duì)列（（Work Queue）
1.3.3、SpringAMQP 實(shí)現(xiàn)廣播交換機(jī)消息隊(duì)列（Fanout Exchange）
1.3.4、Spring AMQP 實(shí)現(xiàn)路由交換機(jī)消息隊(duì)列（Direct Exchange）
1.3.5、SpringAMQP 實(shí)現(xiàn)主題交換機(jī)消息模型（Topic Exchange）
一、SpringAMQP
1.1、概念
什么是 AMQP?
全稱?Advanced Message Queuing Protocol，是用于在應(yīng)用程序之間傳遞業(yè)務(wù)消息的開放標(biāo)準(zhǔn)。該協(xié)議與語言和平臺(tái)無關(guān)，更符合微服務(wù)中獨(dú)立性的要求。
什么是 SpringAMQP?
Spring AMQP是基于AMQP協(xié)議定義的一套API規(guī)范，提供了模板用來發(fā)送和接收消息。包含兩部分，其中spring-amqp是基礎(chǔ)抽象，spring-rabbit是底層的默認(rèn)實(shí)現(xiàn)。
1.2、前置知識(shí)（實(shí)現(xiàn)案例前必看?。?/h3>
1.2.1、創(chuàng)建隊(duì)列
實(shí)際開發(fā)中，有兩種方式創(chuàng)建交換機(jī)和隊(duì)列，其中 Bean 方式用的最多，也是最推薦的，還有一種最后講.
隊(duì)列的創(chuàng)建，源碼中提供了如下幾種方法：
- name：隊(duì)列名.
- durable：是否持久化.
- exclusive：是否獨(dú)占.??true表示隊(duì)列只能被一個(gè)消費(fèi)者使用, false表示大家都能用.
- autoDelete：自動(dòng)刪除，為 true 表示沒有人使用以后自動(dòng)刪除.
- arguments：擴(kuò)展參數(shù)，自定義選項(xiàng)，例如實(shí)現(xiàn)，隊(duì)列過期、隊(duì)列消息過期、死信隊(duì)列...
例如，創(chuàng)建名為 simple.queue 的隊(duì)列，并且支持持久化：在 Configuration 層中使用 @Bean 注入隊(duì)列即可（@Configuration 注解不能少）.
?
@Configuration
public class MqConfig {
/**
* 創(chuàng)建隊(duì)列
* @return
*/
@Bean
public Queue simpleQueue() {
return new Queue("simple.queue", true);
}
}
?Ps：建議將來開發(fā)的時(shí)候，將設(shè)置隊(duì)列名、交換機(jī)名都寫成固定的常量（static final）
1.2.2、創(chuàng)建交換機(jī)
a）直接交換機(jī)和扇出交換機(jī)
直接交換機(jī)的創(chuàng)建，源碼中提供了如下幾種方法：
扇出交換機(jī)和主題交換機(jī)也是如此：
- name：交換機(jī)名.
- durable：是否持久化.
- autoDelete：自動(dòng)刪除，若當(dāng)前交換機(jī)沒人使用了，就會(huì)自動(dòng)刪除.
- arguments：表示創(chuàng)建交換機(jī)時(shí)可以指定一些額外的選項(xiàng)，例如延時(shí)消息、過期時(shí)間、死信交換機(jī).
例如，創(chuàng)建名為? direct 的交換機(jī)，支持持久化和自動(dòng)刪除.
@Configuration
public class MqConfig {
/**
* 創(chuàng)建直接交換機(jī)
* @return
*/
@Bean
public DirectExchange directExchange() {
return new DirectExchange("direct", true, true);
}
}
1.2.3、創(chuàng)建綁定
通過 BindingBuilder 對(duì)象就可以將交換機(jī)和隊(duì)列綁定，綁定方式是 BindingBuilder.bind(調(diào)用創(chuàng)建隊(duì)列方法).to(調(diào)用創(chuàng)建交換機(jī)方法).with(bindingKey).
@Configuration
public class MqConfig {
/**
* 創(chuàng)建隊(duì)列
* @return
*/
@Bean
public Queue simpleQueue() {
return new Queue("simple.queue", true);
}
/**
* 創(chuàng)建直接交換機(jī)
* @return
*/
@Bean
public DirectExchange directExchange() {
return new DirectExchange("direct", true, true);
}
/**
* 創(chuàng)建綁定
* @return
*/
@Bean
public Binding simpleBinding() {
return BindingBuilder.bind(simpleQueue()).to(directExchange()).with("simple");
}
}
Ps：如果是 fanout 扇出交換機(jī)則不用通過 with 指定 bindingKey.
1.2.4、@RabbitListener 注解
a）情況一：queue 存在
如果 queue 隊(duì)列存在，以下使用就可以通過 @RabbitListener 注解監(jiān)聽隊(duì)列消息，成為消費(fèi)者.
用法：直接指定注解中的queues參數(shù)即可，參數(shù)值為對(duì)列名(queueName)
例如如下
@RabbitListener(queues = "simple.queue")
public void SimpleConsumer(String msg) {
System.out.println("消費(fèi)者收到消息: " + msg);
}
Ps：當(dāng) queue 的 autoDelete 屬性為 false 時(shí)，這種使用情況還是比較合適了；但是，當(dāng)這個(gè)屬性為 true 時(shí)，沒有消費(fèi)者隊(duì)列就會(huì)自動(dòng)刪除了，可能會(huì)引發(fā)異常！
b）情況二：queue 不存在?
如果隊(duì)列存在，以下使用就可以通過 @RabbitListener 注解創(chuàng)建交換機(jī)、隊(duì)列、綁定關(guān)系.
例如，
@RabbitListener(bindings = @QueueBinding(
value = @Queue("queue1"),
exchange = @Exchange(name = "exchange1", type = ExchangeTypes.TOPIC),
key = "test"
))
public void testListener(String msg) {
System.out.println("消費(fèi)者收到消息: " + msg);
}
- value: @Queue 注解，用于聲明隊(duì)列，value 為 queueName, durable 表示隊(duì)列是否持久化, autoDelete 表示沒有消費(fèi)者之后隊(duì)列是否自動(dòng)刪除
- exchange: @Exchange 注解，用于聲明 exchange， type 就是交換機(jī)類型.
- key: 在 topic 方式下，這個(gè)就是我們熟知的 routingKey
1.2.5、為什么更建議使用 @Bean 注解創(chuàng)建，而不是 @RabbitListener 注解創(chuàng)建？
建議使用 @Bean 而不是@RabbitListener的原因如下：
- 使用 @Bean 創(chuàng)建相比于?@RabbitListener 實(shí)現(xiàn)了配置和業(yè)務(wù)上的解耦合（隊(duì)列、交換機(jī)、綁定的創(chuàng)建 和 業(yè)務(wù)代碼 解耦合）.
- @Bean可以更好地控制RabbitMQ監(jiān)聽器的創(chuàng)建和銷毀，而@RabbitListener則需要在運(yùn)行時(shí)動(dòng)態(tài)創(chuàng)建和銷毀監(jiān)聽器，這可能會(huì)導(dǎo)致一些性能問題和內(nèi)存泄漏問題。
- 使用@Bean還可以更好地管理RabbitMQ連接和通道，從而提高應(yīng)用程序的可靠性和性能。
1.3、案例實(shí)現(xiàn)
1.3.1、SpringAMQP 實(shí)現(xiàn)基礎(chǔ)消息隊(duì)列（BasicQueue）
在mq上創(chuàng)建隊(duì)列：
Configuration 這一層創(chuàng)建隊(duì)列，否則不生效.
@Configuration
public class MqConfig {
/**
* 創(chuàng)建隊(duì)列
* @return
*/
@Bean
public Queue simpleQueue() {
return new Queue("simple.queue", true);
}
}
.?
服務(wù)生產(chǎn)者?publisher：
a）由于生產(chǎn)者和消費(fèi)者服務(wù)都需要 amqp 依賴，因此這里直接將依賴放到父工程中
<!--AMQP依賴，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
b）在 publisher （消息發(fā)送）服務(wù)中編寫application.yml，添加mq連接信息：
spring:
rabbitmq:
host: 193.168.150.185 # 主機(jī)名
port: 5672 # 端口
virtual-host: / # 虛擬主機(jī)
username: root # 用戶名
password: 1234 # 密碼Ps：云服務(wù)器記得放行 5672 端口?。?！
c）在publisher服務(wù)中新建一個(gè)測(cè)試類，編寫測(cè)試方法：
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSendMessageSimpleQueue() {
String queueName = "simple.queue";
String message = "hello, spring amqp!";
rabbitTemplate.convertAndSend(queueName, message);
}
}
以上可得知 SpringAMQP 發(fā)送消息需要實(shí)現(xiàn)以下幾點(diǎn)：?
- 引入 amqp 的 starter 依賴；
- 配置 RabbitMQ 地址；
- 利用 RabbitTemplate 的 convertAndSend 方法來指定隊(duì)列發(fā)送消息。
服務(wù)消費(fèi)者 consumer：
a）在consumer服務(wù)中編寫application.yml，添加mq連接信息：
spring:
rabbitmq:
host: 193.168.150.185 # 主機(jī)名
port: 5672 # 端口
virtual-host: / # 虛擬主機(jī)
username: root # 用戶名
password: 1234 # 密碼b）在consumer服務(wù)中新建一個(gè)類，編寫消費(fèi)邏輯：
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void ListenQueue(String msg) {
System.out.println("消費(fèi)者接收到 simple.queue 的消息:" + msg);
}
}
以上可得知 SpringAMQP 接收消息需要實(shí)現(xiàn)以下幾點(diǎn)：?
- 引入amqp的starter依賴
- 配置RabbitMQ地址
- 定義類，添加@Component注解
- 類中聲明方法，添加@RabbitListener注解，方法參數(shù)就時(shí)消息
Ps：消息一旦消費(fèi)就會(huì)從隊(duì)列刪除，RabbitMQ沒有消息回溯功能
最終執(zhí)行效果如下：
1.3.2、SpringAMQP 實(shí)現(xiàn)工作消息隊(duì)列（（Work Queue）
Work queue工作隊(duì)列，可以提高消息處理速度，避免隊(duì)列消息堆積；
遇到的問題：當(dāng)消息進(jìn)入 queue 時(shí)，消費(fèi)者1 和 消費(fèi)者2 都會(huì)先做一個(gè)操作叫做“消息預(yù)取”，也就是先取消息，但不執(zhí)行操作，等從 queue 中拿完（如果不設(shè)置，默認(rèn)是無限次拿）了以后再執(zhí)行具體的消費(fèi)操作，但這樣會(huì)導(dǎo)致一個(gè)問題——當(dāng) 消費(fèi)者1 的消費(fèi)處理速度是?消費(fèi)者2 處理速度的兩倍時(shí)，但是由于“消息預(yù)取”，他們兩都會(huì)取到相等的消息數(shù)量，因此就會(huì)導(dǎo)致 消費(fèi)者1 提前處理完，消費(fèi)者2 還有很多沒有處理的情況，降低了總體處理的效率。
解決辦法：為了解決這種問題的發(fā)生，也可以通過修改消息預(yù)取的數(shù)量，設(shè)置為 1 ，就是拿一個(gè)處理完了以后再取下一個(gè)，就可以避免以上問題（設(shè)置為1，就是根據(jù)自身處理能力，拿去相應(yīng)的數(shù)據(jù)）。
具體實(shí)現(xiàn)如下
a）在 Configuration 中創(chuàng)建好隊(duì)列
@Configuration
public class MqConfig {
/**
* 創(chuàng)建隊(duì)列
* @return
*/
@Bean
public Queue simpleQueue() {
return new Queue("simple.queue", true);
}
}
b）在?publisher?服務(wù)中添加一個(gè)測(cè)試方法，循環(huán)發(fā)送?50?條消息到?simple.queue?隊(duì)列（控制時(shí)間為1秒內(nèi)發(fā)送完）
@Test
public void testSendMessageWorkQueue() throws InterruptedException {
String queueName = "simple.queue";
String message = "hello, workQueue!";
for(int i = 0; i < 50; i++) {
rabbitTemplate.convertAndSend(queueName, message + i);
Thread.sleep(20);
}
}
c）在consumer服務(wù)中創(chuàng)建兩個(gè)消費(fèi)者，也監(jiān)聽simple.queue：
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void ListenQueue(String msg) throws InterruptedException {
System.err.println("消費(fèi)者接收到 simple.queue 的消息:" + msg + LocalDate.now());
Thread.sleep(100);
}
@RabbitListener(queues = "simple.queue")
public void ListenQueue2(String msg) throws InterruptedException {
System.out.println("消費(fèi)者接收到 simple.queue 的消息:" + msg + LocalDate.now());
Thread.sleep(25);
}
}
d）消費(fèi)預(yù)取限制：修改application.yml文件，設(shè)置preFetch這個(gè)值，可以控制預(yù)取消息的上限
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 170.193.116.135
port: 5672
username: root
password: 1234
virtual-host: /
listener:
simple:
prefetch: 1
執(zhí)行結(jié)果如下：?
1.3.3、SpringAMQP 實(shí)現(xiàn)廣播交換機(jī)消息隊(duì)列（Fanout Exchange）
發(fā)布訂閱模式與之前案例的區(qū)別就是允許將同一消息發(fā)送給多個(gè)消費(fèi)者。實(shí)現(xiàn)方式是加入了exchange（交換機(jī)）。
常見exchange類型包括：
1. Fanout：廣播
2. Direct：路由
3. Topic：話題
這里我們先來理解一下廣播消息隊(duì)列模型~
Fanout Exchange：會(huì)將接收到的消息廣播到每一個(gè)跟其綁定的queue，最后交給對(duì)應(yīng)的消費(fèi)者，值得注意的是，exchange負(fù)責(zé)消息路由，而不是存儲(chǔ)，路由失敗則消息丟失。
具體實(shí)現(xiàn)如下：
a）在consumer服務(wù)創(chuàng)建一個(gè)類，添加@Configuration注解，創(chuàng)建隊(duì)列和交換機(jī)，并聲明FanoutExchange、Queue和綁定關(guān)系對(duì)象?Binding ，綁定方式是 BindingBuilder.bind(隊(duì)列).to(交換機(jī))
package cn.itcast.mq.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class FanoutConfig {
//itcast.fanout
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("itcast.fanout");
}
//itcast.queue1
@Bean
public Queue fanoutQueue1() {
return new Queue("fanout.queue1");
}
//綁定隊(duì)列 1 到交換機(jī)
@Bean
public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
return BindingBuilder
.bind(fanoutQueue1)
.to(fanoutExchange);
}
//itcast.queue2
@Bean
public Queue fanoutQueue2() {
return new Queue("fanout.queue2");
}
//綁定隊(duì)列 2 到交換機(jī)
@Bean
public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
return BindingBuilder
.bind(fanoutQueue2)
.to(fanoutExchange);
}
}
b）運(yùn)行以后可以通過 RabbitMQ 客戶端看到以下綁定情況：
?
c）添加測(cè)試方法
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAMQPTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testFanoutMessage() {
String exchangeName = "itcast.fanout";
String message = "hello! fanout";
rabbitTemplate.convertAndSend(exchangeName, "", message);
}
}
d）這里一個(gè)交換機(jī)綁定了兩個(gè)隊(duì)列，因此運(yùn)行結(jié)果如下?：
1.3.4、Spring AMQP 實(shí)現(xiàn)路由交換機(jī)消息隊(duì)列（Direct Exchange）
?Direct Exchange 會(huì)將接收到的消息根據(jù)規(guī)則路由到指定的Queue，因此稱為路由模式（routes），通常一個(gè) Queue 都會(huì)與 Exchange 設(shè)置一個(gè) BindingKey，發(fā)布者發(fā)送消息時(shí)會(huì)先指定 RoutingKey，這時(shí)交換機(jī)就會(huì)將消息路由到 BindingKey?與消息?RoutingKey?一致的隊(duì)列。
具體實(shí)現(xiàn)如下：
a）在consumer服務(wù)中，編寫兩個(gè)消費(fèi)者方法，分別監(jiān)聽direct.queue1和direct.queue2，并利用@RabbitListener聲明Queue、Exchange、BoutingKey
Ps：之氣我們使用 Bean 的方式創(chuàng)建交換機(jī)和隊(duì)列，實(shí)際上也可以通過 @RabbitListener 注解去聲明.? 這就要求我們必須要在 @RabbitListener 中通過 @QueueBinding 指定綁定關(guān)系，并通過 @Queue 和 @Exchange 去創(chuàng)建隊(duì)列和交換機(jī)
但建議將來微服務(wù)開發(fā)的時(shí)候都是用 Bean 的方式創(chuàng)建交換機(jī)和隊(duì)列，將業(yè)務(wù)代碼和配置代碼解耦合.
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"), //綁定隊(duì)列
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT), //綁定交換機(jī)，type 可以不寫，默認(rèn)為 Direct
key = {"java", "C++"} // Bindingkey
))
public void listenDirectQueue1(String msg) {
System.out.println("消費(fèi)者 1 收到 Direct 消息：" + msg);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"), //綁定隊(duì)列
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT), //綁定交換機(jī)，type 可以不寫，默認(rèn)為 Direct
key = {"java", "GO"} // Bindingkey
))
public void listenDirectQueue2(String msg) {
System.out.println("消費(fèi)者 2 收到 Direct 消息：" + msg);
}
b）運(yùn)行后可以觀察綁定情況：
c）編寫測(cè)試方法：
@Test
public void testDirectMessage() {
String exchangeName = "itcast.direct";
String message = "hello! direct";
rabbitTemplate.convertAndSend(exchangeName, "C++", message);
}
d）運(yùn)行結(jié)果如下：
1.3.5、SpringAMQP 實(shí)現(xiàn)主題交換機(jī)消息模型（Topic Exchange）
TopicExchange?與?DirectExchange 十分類似，區(qū)別在于routingKey必須是多個(gè)單詞的列表，并且以 . 分割。
Queue與Exchange指定BindingKey時(shí)可以使用以下通配符：
#：代指0個(gè)或多個(gè)單詞
*：代指一個(gè)單詞
例如： china.# 就相當(dāng)于 china.new 或者 china.difa 或者 china.alhglag ......等等一切情況
具體實(shí)現(xiàn)如下（這里不具體說明了，除了通配符差異，用法跟 Direct 消息隊(duì)列一模一樣~）：
@Component
public class SpringRabbitListener {
@RabbitListener(bindings = @QueueBinding(
value = @Queue("topic.queue1"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "china.#"
))
public void listenTopicQueue1(String msg) {
System.out.println("消費(fèi)者 1 收到 topic 消息：" + msg);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue("topic.queue2"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "#.good"
))
public void listenTopicQueue2(String msg) {
System.out.println("消費(fèi)者 2 收到 topic 消息：" + msg);
}
}
@Test
public void testTopicMessage() {
String exchangeName = "itcast.topic";
String message = "hello! topic";
rabbitTemplate.convertAndSend(exchangeName, "china.good", message);
}
運(yùn)行結(jié)果：
文章來源：http://www.zghlxwxcb.cn/news/detail-848877.html
文章來源地址http://www.zghlxwxcb.cn/news/detail-848877.html
到了這里，關(guān)于RabbitMQ - 基于 SpringAMQP 帶你實(shí)現(xiàn)五種消息隊(duì)列模型的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
