SpringBoot2實(shí)現(xiàn)MessageQueue消息隊(duì)列
更新時(shí)間：2023年04月25日 09:47:34 作者：yjx23332
本文主要介紹了 SpringBoot2實(shí)現(xiàn)MessageQueue消息隊(duì)列,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
什么是消息隊(duì)列
通常說的消息隊(duì)列，簡稱MQ（Message Queue），指的就是消息中間件。簡單理解為一個(gè)使用隊(duì)列來通信的組件，本質(zhì)上就是個(gè)轉(zhuǎn)發(fā)器，包含發(fā)消息，存消息，消費(fèi)消息的過程。
一、異步與同步
1.1 同步通訊與異步通訊
- 同步通訊：時(shí)效性強(qiáng)。比如視頻電話，實(shí)時(shí)傳到對(duì)方，同時(shí)對(duì)方出回應(yīng)。
- 異步通訊：比如網(wǎng)絡(luò)聊天，非實(shí)時(shí)反饋的，不會(huì)立即得到結(jié)果，可以之后再回復(fù)。
1.2 同步調(diào)用的問題
微服務(wù)基于Feign的調(diào)用就是同步的方式。
以購物場景為例
但是如果要加業(yè)務(wù)就需要為支付服務(wù)加業(yè)務(wù)，改動(dòng)其代碼，耦合度高。
同時(shí)，同步調(diào)用，要等待服務(wù)結(jié)束后，在進(jìn)行下一個(gè)服務(wù)。支付總耗時(shí)，是支付服務(wù)依次調(diào)用服務(wù)耗時(shí)的時(shí)間和，耗時(shí)過長。
此外，如果倉儲(chǔ)服務(wù)掛掉了，支付服務(wù)就會(huì)被卡在那里。當(dāng)過多的支付服務(wù)都卡在那里，于是資源耗盡，支付服務(wù)也掛掉了。
問題：
- 耦合度高
- 性能下降
- 資源浪費(fèi)
- 級(jí)聯(lián)失敗
1.3 異步調(diào)用方案
異步調(diào)用常見的就是事件驅(qū)動(dòng)模式
當(dāng)支付服務(wù)告知了Broker后，就可以繼續(xù)自己的事情了，而不需要等待。
優(yōu)勢：
- 代碼解耦合：不需要改動(dòng)支付服務(wù)，只需要讓服務(wù)訂閱或者取消訂閱Broker即可。
- 耗時(shí)減少了：只計(jì)算支付服務(wù)和通知Broker的時(shí)間。
- 不存在級(jí)聯(lián)失敗的問題，倉儲(chǔ)服務(wù)掛了不再影響支付服務(wù)。
- 流量消峰：當(dāng)流量過大時(shí)，請(qǐng)求排在Broker中，服務(wù)能做幾個(gè)就做幾個(gè)，做不了的就排著。
缺點(diǎn)：
- Broker掛了也會(huì)出問題，依賴于Broker的可靠性，安全性，吞吐能力
- 架構(gòu)復(fù)雜了，業(yè)務(wù)沒有明顯的流程線，不好追蹤管理
二、MQ消息隊(duì)列
在上述的結(jié)構(gòu)中，就是Broker。
常用的MQ有幾種實(shí)現(xiàn)。
| RabbitMQ | ActiveMQ | RocketMQ | Kafaka | |
|---|---|---|---|---|
| 公司/社區(qū) | Rabbit | Apache | 阿里 | Apache |
| 開發(fā)語言 | Erlang | Java | Java | Scala&Java |
| 協(xié)議支持 | AMQP、XMPP、SMTP、STOMP | OpenWire、STOMP、REST、XMPP、AMQP | 自定義協(xié)議 | 自定義協(xié)議 |
| 可用性 | 高 | 一般 | 高 | 高 |
| 單機(jī)吞吐量 | 一般 | 差 | 高 | 非常高 |
| 消息延遲 | 微秒級(jí) | 毫秒級(jí) | 毫秒級(jí) | 毫秒以內(nèi) |
| 消息可靠性 | 高 | 一般 | 高 | 一般 |
- 一般中小型公司，用的就是RabbitMQ。
- 如果大型企業(yè)，做深度定制，可以用RocketMQ
- Kafaka則是用于大量數(shù)據(jù)情況下的處理，但安全可靠性相對(duì)較差。
- ActiveMQ是很早的消息隊(duì)列，如今幾乎沒有維護(hù)。
2.1 單機(jī)部署MQ
通過docker部署最簡單，
docker pull rabbitmq:3-management
也可以用命令安裝,這里直接用容器了。
啟動(dòng)信息如下
docker run -e RABBITMQ_DEFAULT_USER=yjx23332 -e RABBITMQ_DEFAULT_PASS=123456 -v mq-plugins:/plugins --name mq --hostname mq1 -p 15672:15672 -p 5672:5672 -d rabbitmq:3-management
-e 為設(shè)置環(huán)境變量
兩個(gè)端口，15672是管理平臺(tái)端口，5672是發(fā)送消息的端口。
記得開放對(duì)應(yīng)端口。如果是騰訊云或者阿里云，也要在購買的服務(wù)器管理頁面打開放行端口。
firewall-cmd --zone=public --add-port=15672/tcp --permanent
firewall-cmd --query-port=15672/tcp 查看某個(gè)端口firewall-cmd --zone=public --list-ports 查看所有
firewall-cmd --zone=public --add-port=5672/tcp --permanent
firewall-cmd --reload
查詢端口是否開放
登陸成功后，即可進(jìn)入以下界面。
我們可在這里為添加用戶和角色
virtualhosts虛擬主機(jī)：對(duì)不用戶進(jìn)行隔離，避免相互影響。
此處可以添加
點(diǎn)擊用戶，可以配置其虛擬主機(jī)權(quán)限等。
此處設(shè)置交換機(jī)
2.2 結(jié)構(gòu)和概念
使用消息隊(duì)列中消息的對(duì)象。我們稱之為消費(fèi)者。
在一個(gè)virtualhost下：
2.3 常見的消息模型
基本消息隊(duì)列BasicQueue：最簡單的實(shí)現(xiàn)
工作消息隊(duì)列WorkQueue：在工作者之間分配任務(wù)
發(fā)布訂閱帶有交換機(jī)，分為：
Fanout Exchange：廣播，發(fā)布訂閱（publish/subscribe）：一次性向讀個(gè)消費(fèi)者發(fā)送消息。
Direct Exchange：路由（Routing）：有選擇的接收消息
Topic Exchange：主題 （Topics）：根據(jù)主題接收消息。
請(qǐng)求回復(fù)模型（RPC）：收到請(qǐng)求然后答復(fù)。
發(fā)布者確認(rèn)模式（Publisher Confirms）：會(huì)讓發(fā)布者知道發(fā)送是否成功。
三、SpringAMQP
3.1 用非自動(dòng)裝配的方式使用消息隊(duì)列
需要在項(xiàng)目中引入AMQP，記得加入父類spring-boot-starter-parent
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
引入Junit方便測試
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
測試一下MQ
我們創(chuàng)建如下兩個(gè)子項(xiàng)目。
為test寫一個(gè)測試用例
package com.yjx23332.mq.helloworld;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class PublisherTest {
@Test
public void testSendMessage() throws IOException, TimeoutException{
//1.建立連接
ConnectionFactory factory = new ConnectionFactory();
//1.2.設(shè)置連接參數(shù)
factory.setHost("IP");//MQ地址設(shè)置
factory.setPort(5672);//端口設(shè)置
factory.setVirtualHost("/");//設(shè)置虛擬主機(jī)
factory.setUsername("賬號(hào)");
factory.setPassword("密碼");
//1.2 建立連接
Connection connection = factory.newConnection();
//2.創(chuàng)建通道
Channel channel = connection.createChannel();
//3.創(chuàng)建消息隊(duì)列
String queueName = "simple.queue";
channel.queueDeclare(queueName,false,false,false,null);
//4.發(fā)送消息
String message = "hello,rabbitmq!";
channel.basicPublish("",queueName,null,message.getBytes());
System.out.println("已發(fā)送消息：【"+message+"】");
//5.關(guān)閉通道
if(channel != null){
channel.close();
}
if(connection != null){
connection.close();
}
}
}
我們?cè)诎l(fā)送消息前打上斷點(diǎn)，用junit運(yùn)行，就可以看到連接創(chuàng)建和通道創(chuàng)建.。
因?yàn)榘l(fā)完就不管了，因此必須打斷點(diǎn)，才看得到連接和通道。
完成后可以看到隊(duì)列中
接下來我們處理消息,基本一樣，只需要修改幾個(gè)部分。
注意我們沒有關(guān)閉連接，因?yàn)樵跇I(yè)務(wù)中，要一直處理。
package com.yjx23332.mq.helloworld;
import com.rabbitmq.client.*;
import org.junit.Test;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ConsumerTest {
@Test
public static void main(String[] args) throws IOException, TimeoutException{
//1.建立連接
ConnectionFactory factory = new ConnectionFactory();
//1.2.設(shè)置連接參數(shù)
factory.setHost("101.43.65.53");//MQ地址設(shè)置
factory.setPort(5672);//端口設(shè)置
factory.setVirtualHost("/");//設(shè)置虛擬主機(jī)
factory.setUsername("yjx23332");
factory.setPassword("123456");
//1.2 建立連接
Connection connection = factory.newConnection();
//2.創(chuàng)建通道
Channel channel = connection.createChannel();
//3.創(chuàng)建消息隊(duì)列
//為什么這里也要?jiǎng)?chuàng)建？避免消費(fèi)者先執(zhí)行，還沒有隊(duì)列。同時(shí)，相同的隊(duì)列創(chuàng)建重復(fù)執(zhí)行沒有影響。
String queueName = "simple.queue";
channel.queueDeclare(queueName,false,false,false,null);
//4.處理消息
String message = "hello,rabbitmq!";
//DefaultConsumer 是回調(diào)函數(shù)，一旦有消息，異步處理
channel.basicConsume(queueName,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{
System.out.println("接收到消息：【"+ new String (body)+"】");
}
});
System.out.println("####################等待接收消息##################");
//
// //5.關(guān)閉通道
// if(channel != null){
// channel.close();
// }
// if(connection != null){
// connection.close();
// }
}
}
結(jié)果如下
可以看到，因?yàn)榛卣{(diào)的原因，后面的輸出先執(zhí)行
隊(duì)列中消息處理完畢
3.2 SpringAMQP介紹
AMQP：Advanced Message Queuing Protocol：高級(jí)消息隊(duì)列協(xié)議。于應(yīng)用程序之間傳遞業(yè)務(wù)消息的開放標(biāo)準(zhǔn)。
Spring AMQP：基于AMQP協(xié)議的一套API規(guī)范，提供模板來發(fā)送和接收消息。其中Spring-amqp是基礎(chǔ)抽象，Spring-rabbit是底層的默認(rèn)實(shí)現(xiàn)。可參考Spring AMQP官網(wǎng)。
3.3 基礎(chǔ)消息隊(duì)列功能使用
導(dǎo)入依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
為了方便測試，我們引入SpringBoot單元測試
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
然后準(zhǔn)備一個(gè)yml文件，配置和之前用代碼寫得相似。
spring:
rabbitmq:
host:
port: 5672
virtual-host: /
username:
password:
我們直接走單元測試，這里就不創(chuàng)建一個(gè)隊(duì)列了，直接放消息。
package com.yjx23332.mq.helloworld;
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;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
@SpringBootTest
@RunWith(SpringRunner.class)
public class PublisherTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() throws IOException, TimeoutException{
rabbitTemplate.convertAndSend("simple.queue","hello,spring amqp");
}
}
接下來為消費(fèi)者建一個(gè)監(jiān)聽器(記得配置yml文件)
package com.yjx23332.mq.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException{
System.out.println("spring 消費(fèi)者接收到消息:【"+msg+"】");
}
}
消息一旦消費(fèi)，就會(huì)被移除，Rabbit MQ不存在回溯功能。
3.4 工作隊(duì)列的配置
一個(gè)隊(duì)列綁定多個(gè)消費(fèi)者。
我們準(zhǔn)備發(fā)送50條消息
package com.yjx23332.mq.helloworld;
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;
@SpringBootTest
@RunWith(SpringRunner.class)
public class PublisherTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() throws InterruptedException{
for(int i = 0;i < 50;i++){
rabbitTemplate.convertAndSend("simple.queue","hello,spring amqp___" + i);
Thread.sleep(20);
}
}
}修改消費(fèi)者
package com.yjx23332.mq.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalTime;
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException{
System.out.println("spring 消費(fèi)者接收到消息:【"+msg+"】" + LocalTime.now());
Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage2(String msg) throws InterruptedException{
System.err.println("spring 消費(fèi)者接收到消息:【"+msg+"】"+ LocalTime.now());
Thread.sleep(200);
}
}從結(jié)果會(huì)發(fā)現(xiàn)處理總時(shí)長超過了1秒達(dá)到了5秒，查看輸出會(huì)發(fā)現(xiàn)消息被平均分配給了兩個(gè)。一個(gè)處理偶數(shù)，一個(gè)處理奇數(shù)。但由于處理速度不同，因此處理總時(shí)長超過了1秒。
這里是因?yàn)橄M(fèi)預(yù)取導(dǎo)致的，在執(zhí)行前會(huì)提前把消息從隊(duì)列拿出，然后各自處理。
但我們希望的是，做的快的多做，做的慢的少做。
因此我們可以修改yml文件：
spring:
rabbitmq:
host:
port: 5672
virtual-host: /
username:
password:
listener:
simple:
prefetch: 1 # 每次只能獲取幾條消息，執(zhí)行完了再取下一條，默認(rèn)是無限
重啟后再次執(zhí)行就會(huì)發(fā)現(xiàn)正常了。
3.5 發(fā)布與訂閱模式
我們需要將同一消息發(fā)送給多個(gè)消費(fèi)者。需要加入交換機(jī)來實(shí)現(xiàn)。注意，交換機(jī)只負(fù)責(zé)消費(fèi)路由，但不存儲(chǔ)消息，丟失一概不負(fù)責(zé)。
3.5.1 SpringAMQP交換機(jī)類
3.5.2 Fanout Exchange
我們?cè)赾onsumer服務(wù)中聲明Exchange、Queue、Binding.
package com.yjx23332.mq.confg;
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 {
//聲明FanoutExchange交換機(jī)
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("yjx23332.fanout");
}
//聲明一個(gè)隊(duì)列
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
//綁定隊(duì)列和交換機(jī)
@Bean
public Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
//聲明第二個(gè)隊(duì)列
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
//綁定第二個(gè)隊(duì)列和交換機(jī)
@Bean
public Binding bindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
}
}
運(yùn)行后，會(huì)看到：
修改監(jiān)聽器
package com.yjx23332.mq.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class MQlistener {
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg){
System.out.println("spring 消費(fèi)者接收q1到消息:【"+msg+"】");
}
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg){
System.err.println("spring 消費(fèi)者接收到q2消息:【"+msg+"】");
}
}
我們?cè)傩薷膒ublisher的測試代碼
package com.yjx23332.mq;
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;
@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testFanoutExchange(){
String exchangeName = "yjx23332.fanout";
String message = "hello world!";
rabbitTemplate.convertAndSend(exchangeName,"",message);
}
}
啟動(dòng)
3.5.3 DirectExchange
將接收到的消息根據(jù)規(guī)則路由到指定的Queue，因此稱為路由模式（routes）。
- 每一個(gè)Queue都與Exchange設(shè)置一個(gè)BindingKey
- 發(fā)布者發(fā)送消息時(shí)，指定消息的RoutingKey
- Exchange將消息路由到BindingKey與消息RoutingKey一致的隊(duì)列
- 一個(gè)隊(duì)列可以指定多個(gè)BindingKey，且隊(duì)列之間的BindingKey可以重復(fù)
由于基于Config創(chuàng)建隊(duì)列交換機(jī)的方式很麻煩，我們用新的方式聲明交換機(jī)、隊(duì)列。
刪除上一節(jié)我們?cè)赾onfig中的聲明代碼。
然后在listener中進(jìn)行
package com.yjx23332.mq.listener;
import org.springframework.amqp.core.ExchangeTypes;
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;
@Component
public class MQlistener {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(value = "yjx23332.direct",type = ExchangeTypes.DIRECT),
key = {"red","blue"} //bindingkey
))
public void listenDirectQueue1(String msg){
System.out.println("spring 消費(fèi)者接收q1到消息:【"+msg+"】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(value = "yjx23332.direct" , type = ExchangeTypes.DIRECT),
key = {"red","yellow"}
))
public void listenDirectQueue2(String msg) {
System.err.println("spring 消費(fèi)者接收到q2消息:【"+msg+"】");
}
}運(yùn)行后，我們可以看到
接下來，我們修改Test代碼
package com.yjx23332.mq;
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;
@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testDirectExchange(){
String exchangeName = "yjx23332.direct";
rabbitTemplate.convertAndSend(exchangeName,"red","hello red");
rabbitTemplate.convertAndSend(exchangeName,"blue","hello blue");
rabbitTemplate.convertAndSend(exchangeName,"yellow","hello yellow");
}
}
3.5.4 TopicExchange
與DirectExchange類似，但是它的routingKey必須是多個(gè)單詞表，并用’.'分割。
當(dāng)隊(duì)列與交換機(jī)綁定時(shí)，可以使用通配符。避免當(dāng)bindkey過多導(dǎo)致的麻煩。
#：代表0個(gè)或多個(gè)單詞
*：代指一個(gè)單詞
比如
China.news
Japan.news
就可以用 #.news
同理
China.weather
China.news
就可以用 China.#
我們沿用上一節(jié)的代碼,做一點(diǎn)修改即可
package com.yjx23332.mq.listener;
import org.springframework.amqp.core.ExchangeTypes;
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;
@Component
public class MQlistener {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue"),
exchange = @Exchange(value = "yjx23332.topic",type = ExchangeTypes.TOPIC),
key = {"China.#"} //bindingkey
))
public void listenTopicQueue1(String msg){
System.out.println("spring 消費(fèi)者接收q1到消息:【"+msg+"】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(value = "yjx23332.topic",type = ExchangeTypes.TOPIC),
key = {"#.news"}
))
public void listenTopicQueue2(String msg) {
System.err.println("spring 消費(fèi)者接收到q2消息:【"+msg+"】");
}
}重啟后，可看到
修改Test代碼
package com.yjx23332.mq;
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;
@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testDirectExchange(){
String exchangeName = "yjx23332.topic";
rabbitTemplate.convertAndSend(exchangeName,"China.news","江蘇地表最高溫度將達(dá)到72攝氏度");
rabbitTemplate.convertAndSend(exchangeName,"China.weather","未來溫度仍將升高");
rabbitTemplate.convertAndSend(exchangeName,"Japan.news","安培中槍");
}
}
3.6 消息轉(zhuǎn)換器
在發(fā)送中，我們接收消息的類型是Object。SpringAMQP會(huì)幫我們序列化后變?yōu)樽止?jié)發(fā)送。
用默認(rèn)JDK的序列化ObjectOutputStream是沒有問題的，但是中間過程是亂碼，我們這里改用JSON方式的序列化，這樣在消息隊(duì)列中查看也是正常的。
默認(rèn)JDK的消息信息：
接下來我們配置消息轉(zhuǎn)換。
我們先在消費(fèi)者聲明一個(gè)queue，并設(shè)置處理方式
package com.yjx23332.mq.listener;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class MQlistener {
@RabbitListener(queuesToDeclare = @Queue("object.queue"))
public void listenObjectQueue(String msg){
System.out.println("spring 消費(fèi)者接收到Object消息:【"+msg+"】");
}
}我們?yōu)榘l(fā)送類引入依賴并編寫配置
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-yaml</artifactId>
</dependency>
覆蓋默認(rèn)的消息轉(zhuǎn)換。
package com.yjx23332.mq.config;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MessageConverterConfig {
@Bean
public MessageConverter jsonMessageConverter(){
return new Jackson2JsonMessageConverter();
}
}
隨后修改Test
package com.yjx23332.mq;
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;
import java.util.HashMap;
import java.util.Map;
@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testObjectQueue(){
String queueName = "object.queue";
Map<String,Object> msg = new HashMap<>();
msg.put("name","yjx23332");
msg.put("age",21);
rabbitTemplate.convertAndSend(queueName,msg);
}
}
結(jié)果如下：
這時(shí)消息不再是亂碼
我們?cè)跒橄M(fèi)者配置轉(zhuǎn)換，并修改監(jiān)聽器。當(dāng)然，如果我們?cè)趦蛇叾疾慌渲孟⑥D(zhuǎn)換器，這里結(jié)果是一樣的。
package com.yjx23332.mq.listener;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
public class MQlistener {
@RabbitListener(queuesToDeclare = @Queue("object.queue"))
public void listenObjectQueue(Map<String,Object> msg){
System.out.println("spring 消費(fèi)者接收到Object消息:【 name = "+msg.get("name")+",age = "+msg.get("age")+"】");
}
}
結(jié)果如下
參考文獻(xiàn)
[1]Spring AMQP官網(wǎng)
[2]黑馬程序員Java微服務(wù)
[3]RabbitMQ官方文檔
到此這篇關(guān)于 SpringBoot2實(shí)現(xiàn)MessageQueue消息隊(duì)列的文章就介紹到這了,更多相關(guān) SpringBoot2 MessageQueue消息隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- SpringBoot整合消息隊(duì)列RabbitMQ
- Springboot整合ActiveMQ實(shí)現(xiàn)消息隊(duì)列的過程淺析
- springboot整合消息隊(duì)列RabbitMQ
- springboot整合redis之消息隊(duì)列
- 詳解SpringBoot集成消息隊(duì)列的案例應(yīng)用
- SpringBoot整合RabbitMQ消息隊(duì)列的完整步驟
- SpringBoot集成Redis實(shí)現(xiàn)消息隊(duì)列的方法
- Springboot整合Active消息隊(duì)列
- SpringBoot集成消息隊(duì)列的項(xiàng)目實(shí)踐
相關(guān)文章
Spring?@Configuration?proxyBeanMethods=false問題
這篇文章主要介紹了Spring?@Configuration?proxyBeanMethods=false問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-04-04
Java多線程環(huán)境下SimpleDateFormat類安全轉(zhuǎn)換
這篇文章主要介紹了Java多線程環(huán)境下SimpleDateFormat類安全轉(zhuǎn)換，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-02-02
java學(xué)生信息管理系統(tǒng)設(shè)計(jì)
這篇文章主要為大家詳細(xì)介紹了java學(xué)生信息管理系統(tǒng)設(shè)計(jì)，學(xué)生信息添加進(jìn)入數(shù)據(jù)庫的事務(wù),具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2016-11-11
項(xiàng)目為什么引入log4j而不是logback代碼
這篇文章主要介紹了項(xiàng)目為什么引入log4j而不是logback代碼,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-10-10
RabbitMQ高級(jí)應(yīng)用之消費(fèi)端限流策略basicQos詳解
這篇文章主要介紹了RabbitMQ高級(jí)應(yīng)用之消費(fèi)端限流策略basicQos詳解,高并發(fā)情況下,隊(duì)列里面一瞬間就就積累了上萬條數(shù)據(jù),但是消費(fèi)者無法同時(shí)處理這么多請(qǐng)求,這種場景下我們就需要對(duì)消費(fèi)端進(jìn)行限流,需要的朋友可以參考下2023-08-08
SpringBoot+shardingsphere實(shí)現(xiàn)按月分表功能教程
這篇文章主要介紹了SpringBoot+shardingsphere實(shí)現(xiàn)按月分表功能教程,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-04-04
Spring?JDBC?提供幾種方法和數(shù)據(jù)庫中相應(yīng)的不同的類與接口。我將給出使用JdbcTemplate類框架的經(jīng)典和最受歡迎的方法。本文給大家介紹Spring?JDBC?框架的相關(guān)知識(shí)，感興趣的朋友一起看看吧2021-12-12
這篇文章主要介紹了Java編程中面向字符的輸出流,是Java入門學(xué)習(xí)中的基礎(chǔ)知識(shí),需要的朋友可以參考下2015-10-10
Spring Boot使用RestTemplate消費(fèi)REST服務(wù)的幾個(gè)問題記錄
這篇文章主要介紹了Spring Boot使用RestTemplate消費(fèi)REST服務(wù)的幾個(gè)問題記錄，小編覺得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧2018-06-06
這篇文章主要介紹了IDEA怎么設(shè)置maven配置,本文通過圖文并茂的形式給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2020-10-10