SpringAMQP消息隊列(SpringBoot集成RabbitMQ方式)
一、初始配置
1、導(dǎo)入maven坐標(biāo)
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>2、yml配置
spring:
rabbitmq:
host: 你的rabbitmq的ip
port: 5672
username: guest
password: guest二、基本消息隊列
1、創(chuàng)建隊列
訪問接口：http://localhost:15672，賬號密碼都為guest
進(jìn)入后左下角有Add queue添加隊列，我已添加隊列為MqTest1
2、發(fā)布消息
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads() {
String queue="MqTest1";
String message="message1";
rabbitTemplate.convertAndSend(queue,message);
}
}此時可以看到隊列有一個消息
3、接受消息
package com.rabbitmqdemoconsumer.rabbitmq;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SpringRabbitLeistener {
@RabbitListener(queues = "MqTest1")
public void listenSimpleQueueMessage(String msg){
System.out.println("接收到的消息："+msg);
}
}此時控制臺輸出接收到的消息
三、工作消息隊列（Work Queue）
可以提高消息處理速度，避免隊列消息堆積
1、發(fā)布消息
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads() {
String queue="MqTest1";
String message="message1";
for (int i=0;i<10;i++){
rabbitTemplate.convertAndSend(queue,message);
}
}
}此時隊列有10條消息
2、接受消息
package com.rabbitmqdemoconsumer.rabbitmq;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SpringRabbitLeistener {
@RabbitListener(queues = "MqTest1")
public void listenSimpleQueueMessage1(String msg){
System.out.println("consume1接收到的消息："+msg);
}
@RabbitListener(queues = "MqTest1")
public void listenSimpleQueueMessage2(String msg){
System.out.println("consume2接收到的消息："+msg);
}
}3、控制臺輸出結(jié)果
consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1
consume1接收到的消息：message1
consume2接收到的消息：message1
4、消息預(yù)取問題
但是此時有一個問題就是消息預(yù)取，比如隊列有10條消息，兩個消費(fèi)者各自直接先預(yù)取5個消息，如果一個消費(fèi)者接受消息的速度慢，一個快，就會導(dǎo)致一個消費(fèi)者已經(jīng)完成工作，另一個還在慢慢處理，會造成消息堆積消費(fèi)者身上，要解決這個問題需要在yml文件配置相關(guān)配置
rabbitmq:
host: 43.140.244.236
port: 5672
username: guest
password: guest
virtual-host: /
listener:
simple:
prefetch: 1 #每次只能取一個，處理完才能取下一個消息這樣可以避免消息預(yù)取導(dǎo)致堆積
四、發(fā)布訂閱模式
exchange是交換機(jī)，負(fù)責(zé)消息路由，但不存儲消息，路由失敗則消息丟失
五、發(fā)布訂閱模式之廣播模式（Fanout）
1、Fanout配置類（@Bean聲明）
package com.rabbitmqdemoconsumer.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 FanountConfig {
//交換機(jī)聲明
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("FanountExchange");
}
//聲明隊列1
@Bean
public Queue Fanount_Qeueue1(){
return new Queue("Fanount_Qeueue1");
}
//聲明隊列2
@Bean
public Queue Fanount_Qeueue2(){
return new Queue("Fanount_Qeueue2");
}
//綁定交換機(jī)和隊列
@Bean
public Binding bindingFanount_Qeueue1(Queue Fanount_Qeueue1,FanoutExchange fanoutExchange){
return BindingBuilder.bind(Fanount_Qeueue1).to(fanoutExchange);
}
@Bean
public Binding bindingFanount_Qeueue2(Queue Fanount_Qeueue2,FanoutExchange fanoutExchange){
return BindingBuilder.bind(Fanount_Qeueue2).to(fanoutExchange);
}
}可以看到聲明的隊列
已經(jīng)聲明的交換機(jī)（第一個）
綁定關(guān)系
2、發(fā)送消息
首先發(fā)送10條消息，經(jīng)過交換機(jī)轉(zhuǎn)發(fā)到隊列
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads2() {
String exchange="FanountExchange";
String message="message";
for (int i=0;i<10;i++){
rabbitTemplate.convertAndSend(exchange,"",message);
}
}
}此時可以看到兩個隊列各自有十條消息
3、接受消息
//監(jiān)聽交換機(jī)Fanount_Qeueue1
@RabbitListener(queues = "Fanount_Qeueue1")
public void listenFanountQeueue1(String msg){
System.out.println("Fanount_Qeueue1接收到的消息："+msg);
}
//監(jiān)聽交換機(jī)Fanount_Qeueue2
@RabbitListener(queues = "Fanount_Qeueue2")
public void listenFanountQeueue2(String msg){
System.out.println("Fanount_Qeueue2接收到的消息："+msg);
}控制臺結(jié)果如下（共發(fā)送20條，每個隊列10條）
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue1接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue2接收到的消息：message
Fanount_Qeueue2接收到的消息：message
六、發(fā)布訂閱模式之路由模式（Direct）
會將消息根據(jù)規(guī)則路由到指定的隊列
1、聲明（基于@RabbitListener聲明）
package com.rabbitmqdemoconsumer.rabbitmq;
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 SpringRabbitLeistener {
/**
* 綁定交換機(jī)和隊列，并為key賦值
* @param msg
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "DirectQueue1"),
exchange = @Exchange(name = "DirectExchange",type = ExchangeTypes.DIRECT),
key = {"red","blue"}
))
public void listenDirectQueue1(String msg){
System.out.println("listenDirectQueue1接收到的消息："+msg);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "DirectQueue2"),
exchange = @Exchange(name = "DirectExchange",type = ExchangeTypes.DIRECT),
key = {"red","yellow"}
))
public void listenDirectQueue2(String msg){
System.out.println("listenDirectQueue2接收到的消息："+msg);
}
}此時可以看到聲明的隊列
聲明的交換機(jī)（第一個）
綁定關(guān)系
2、發(fā)送給blue
發(fā)送消息
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads2() {
String exchange="DirectExchange";
String message="HelloWorld";
for (int i=0;i<10;i++){
rabbitTemplate.convertAndSend(exchange,"blue",message);
}
}
}接收消息
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
3、發(fā)送給red
發(fā)送消息
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads2() {
String exchange="DirectExchange";
String message="HelloWorld";
for (int i=0;i<10;i++){
rabbitTemplate.convertAndSend(exchange,"blue",message);
}
}
}接收消息
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
listenDirectQueue2(red,yellow)接收到的消息：HelloWorld
listenDirectQueue1(red,blue)接收到的消息：HelloWorld
七、發(fā)布訂閱模式之廣播模式（Topic）
Queue與Exchange指定BindingKey可以使用通配符：
- #：代指0個或多個單詞
- *：代指一個單詞
比如：
- bindingkey: china.# ->中國的所有消息
- bindingkey: #.weather ->所以國家的天氣
1、聲明
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "TopicQueue1"),
exchange = @Exchange(name = "TopicExchange",type = ExchangeTypes.TOPIC),
key = {"china.#"}
))
public void listenTopicQueue1(String msg){
System.out.println("listenTopicQueue1接收到的消息："+msg);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "TopicQueue2"),
exchange = @Exchange(name = "TopicExchange",type = ExchangeTypes.TOPIC),
key = {"#.news"}
))
public void listenTopicQueue2(String msg){
System.out.println("listenTopicQueue2接收到的消息："+msg);
}隊列
交換機(jī)（第四個）
綁定關(guān)系
2、發(fā)送消息（測試1）
package com.rabbitmqdemo;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads2() {
String exchange="TopicExchange";
String message="HelloWorld";
for (int i=0;i<10;i++){
rabbitTemplate.convertAndSend(exchange,"china.news",message);
}
}
}接收消息
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue2接收到的消息：HelloWorld
3、發(fā)送消息（測試2）
發(fā)送消息
package com.rabbitmqdemo;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class RabbitMQDemoPublishApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads2() {
String exchange="TopicExchange";
String message="HelloWorld";
for (int i=0;i<10;i++){
rabbitTemplate.convertAndSend(exchange,"china.weather",message);
}
}
}接收消息
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
TopicQueue1接收到的消息：HelloWorld
總結(jié)
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
SpringBoot中使用com.alibaba.druid.filter.config.ConfigTools對數(shù)據(jù)庫
這篇文章主要介紹了SpringBoot中使用com.alibaba.druid.filter.config.ConfigTools對數(shù)據(jù)庫密碼加密的方法,本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-01-01
這篇文章主要為大家詳細(xì)介紹了HTTP中g(shù)et和post的區(qū)別，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-08-08
java?集合工具類Collections及Comparable和Comparator排序詳解
這篇文章主要介紹了java集合工具類Collections及Comparable和Comparator排序詳解，文章圍繞主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價值，需要的小伙伴可以參考一下2022-06-06
SpringBoot整合Redis實(shí)現(xiàn)登錄失敗鎖定功能(實(shí)例詳解)
本文我們已經(jīng)探討如何利用Redis來實(shí)現(xiàn)鎖定賬戶的安全措施,以及通過SpringBoot整合Redis實(shí)現(xiàn)了這一功能,感興趣的朋友跟隨小編一起學(xué)習(xí)下吧2024-02-02
Java異常報錯:?java.io.IOException:?Broken?pipe解決方案
這篇文章主要介紹了Java異常報錯:java.io.IOException: Broken pipe的解決方案,java.io.IOException: Broken pipe是Java網(wǎng)絡(luò)編程中最常見的錯誤之一,通常發(fā)生在客戶端和服務(wù)端之間的網(wǎng)絡(luò)連接異常斷開時,需要的朋友可以參考下2025-10-10
Jenkins自動化部署springboot代碼實(shí)例
這篇文章主要介紹了Jenkins自動化部署springboot代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-04-04
這篇文章主要介紹了Java8新特性之默認(rèn)方法(default)淺析,默認(rèn)方法也稱為虛擬擴(kuò)展方法或防護(hù)方法,可以讓我們修改接口而不破壞原來的實(shí)現(xiàn)類的結(jié)構(gòu),需要的朋友可以參考下2014-06-06
JSP文件一般有兩個存放位置,本文主要介紹了JavaWeb項目中JSP訪問的問題解決,具有一定的參考價值,感興趣的可以了解一下2024-01-01
詳解Spring DeferredResult異步操作使用場景
本文主要介紹了Spring DeferredResult異步操作使用場景，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-10-10
SpringBoot+JPA?分頁查詢指定列并返回指定實(shí)體方式
這篇文章主要介紹了SpringBoot+JPA?分頁查詢指定列并返回指定實(shí)體方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12