SpringBoot整合RabbitMQ 實作六種工作模式

消息生産者或者發送者，使用 P 表示：

消息從生産端發送到消費端，一定要通過隊列轉發，使用 queue_name 表示：

消費的消費者或者接收者，使用 C 表示，如果有多個消費者也可以用 C1、C2 表示：

添加 maven 依賴

<dependency>   
  <groupId>org.springframework.boot</groupId>   
  <artifactId>spring-boot-starter-amqp</artifactId>    
  <version>2.2.1.RELEASE</version>
</dependency>

添加 application.yml 配置

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    virtual-host: /
    username: guest
    password: guest

消息生産

生産端發送消息，調用 RabbitTemplate 發送消息，比如：

@Autowired
private RabbitTemplate rabbitTemplate;
public String send() {  
rabbitTemplate.convertAndSend("routingKey","send message");
}

消費消息

消費消息使用隊列監聽注解 @RabbitListener, 添加隊列名稱就能消費發送到隊列上的消息了：

@RabbitListener(queuesToDeclare = @Queue("queue_name"))
public void consume(String message) {

1. 簡單（simple）模式

最簡單的消息發送

生産者是消費者是一一對應，也叫做 點對點模式 , 生産者發送消息經過隊列直接發送給消費者。
生産者和消費者在發送和接收消息時，隻需要指定隊列名稱，而不需要指定 Exchange 交換機。

生産消息:

@GetMapping("/simple-send")
public String simpleSend() { 
 rabbitTemplate.convertAndSend("simple","this is news");  return "ok";
 }

消費消息

@RabbitListener(queuesToDeclare = @Queue("simple"))
public void consume(String message) {
  System.out.println(message);
  }

輸出：

this is news

無需建立交換機和綁定隊列，隻需要比對發送端和消費端的隊列名稱就能成功發送消息。

二工作模式

在多個消費者之間配置設定任務

工作模式和簡單模式差不多，隻需要生産端、消費端、隊列。

不同在于一個生産者、一個隊列對應多個消費者，也就是一對多的關系。

在多個消費者之間配置設定消息（競争消費者模式），類似輪詢發送消息，每個消息都隻發給一個消費者。

生産消息：

@GetMapping("/work-send")
public String simpleSend() { 
 rabbitTemplate.convertAndSend("work","this is news");  return "ok";
 }

消費消息：

@RabbitListener(queuesToDeclare = @Queue("work"))
public void consume(String message) { 
 System.out.println("first:" + message);
 }

@RabbitListener(queuesToDeclare = @Queue("work"))
public void consumeSecond(String message) {  
System.out.println("second:" + message);
}

建立一個生産者，兩個消費者，發送兩條消息，兩個消費者分别接收到消息，輸出：

first:this is newssecond:this is news

兩個消費者，輪流消費消息。類似 nginx 負載均衡。

釋出訂閱模式

一次向多個消費者發送消息

特點:

釋出訂閱類似廣播消息，每個消息可以同時發送給訂閱該消息的消費者，

上圖中的 X 表示交換機，使用的扇形交換機 (fanout), 它将發送的消息發送到所有綁定交換機的隊列。

建立隊列、交換機以及綁定：

@Bean
public FanoutExchange fanoutExchange() { 
 return new FanoutExchange("PUBLISH_SUBSCRIBE_EXCHANGE");
 }

@Beanpublic Queue psFirstQueue() {  return new Queue("psFirstQueue");
}

@Beanpublic Queue psSecondQueue() {  
return new Queue("psSecondQueue");
}

@Bean
public Queue psThirdQueue() {  
return new Queue("psThirdQueue");
}

@Bean
public Binding routingFirstBinding() { 
 return BindingBuilder.bind(psFirstQueue()).to(fanoutExchange());
 }

@Bean
public Binding routingSecondBinding() {  
return BindingBuilder.bind(psSecondQueue()).to(fanoutExchange());}

@Bean
public Binding routingThirdBinding() { 
 return BindingBuilder.bind(psThirdQueue()).to(fanoutExchange());
 }

上面定義一個交換機 fanoutExchange。
分别綁定三個隊列 psFirstQueue、psSecondQueue、psThirdQueue。
隊列綁定交換機不需要 routingKey，直接綁定即可。

生産端：

@GetMapping("/publish-sub-send")
public String publishSubSend() { 
 rabbitTemplate.convertAndSend("PUBLISH_SUBSCRIBE_EXCHANGE", null, "publish/subscribe hello");  return "ok";
 }

無需指定 routingKey, 設定為 null。

消費端：

@RabbitListener(queues = "psFirstQueue")
public void pubsubQueueFirst(String message) { 
 System.out.println("【first】:" + message);
 }

@RabbitListener(queues = "psSecondQueue")
public void pubsubQueueSecond(String message) {  
System.out.println("【second】:" + message);
}

@RabbitListener(queues = "psThirdQueue")
public void pubsubQueueThird(String message) { 
 System.out.println("【third】:" + message);
 }

輸出：

【first】: publish/subscribe hello
【second】: publish/subscribe hello
【third】: publish/subscribe hello

發送一條消息，綁定的隊列都能接收到消息。

路由模式

根據 routingKey 有選擇性的接收消息

特點

每個隊列根據不同 routingKey 綁定交換機
消息發送到交換機後通過 routingKey 發送給特定的隊列，然後傳到消費者消費。
交換由扇形交換機 (fanout) 改成直連交換機 (direct)。

建立隊列、交換機以及綁定：

@Bean
public Queue routingFirstQueue() {    
return new Queue("routingFirstQueue");
}

@Bean
public Queue routingSecondQueue() {  
  return new Queue("routingSecondQueue");
  }
@Bean
public Queue routingThirdQueue() {   
 return new Queue("routingThirdQueue");
}

@Bean
public DirectExchange routingExchange() {    
return new DirectExchange("routingExchange");
}

@Bean
public Binding routingFirstBind() {   
 return BindingBuilder.bind(routingFirstQueue()).to(routingExchange()).with("firstRouting");
 }

@Bean
public Binding routingSecondBind() {   
 return BindingBuilder.bind(routingSecondQueue()).to(routingExchange()).with("secondRouting");
 }
@Bean
public Binding routingThirdBind() {    
return BindingBuilder.bind(routingThirdQueue()).to(routingExchange()).with("thirdRouting");}

建立一個交換機，根據不同的路由規則比對不同的隊列 routingExchange, 根據不同的 routingKey 綁定不同的隊列：
firstRouting 路由鍵綁定 routingFirstQueue 隊列。
secondRouting 路由鍵綁定 routingSecondQueue 隊列。
thirdRouting 路由鍵綁定 routingThirdQueue 隊列。

生産消息:

// 使用不同的routingKey 轉發到不同的隊列    
@GetMapping("/routing-first")
public String routingFirst() {    rabbitTemplate.convertAndSend("routingExchange","firstRouting"," first routing message");    rabbitTemplate.convertAndSend("routingExchange","secondRouting"," second routing message");    rabbitTemplate.convertAndSend("routingExchange","thirdRouting"," third routing message");    return "ok";}

消費消息：

@RabbitListener(queues = "routingFirstQueue")
public void routingFirstListener(String message) 
{    System.out.println("【routing first】" + message);}

@RabbitListener(queues = "routingSecondQueue")
public void routingSecondListener(String message) {  
  System.out.println("【routing second】" + message);}

@RabbitListener(queues = "routingThirdQueue")
public void routingThirdListener(String message) {    
System.out.println("【routing third】" + message);}

輸出：

【routing first】first routing message
【routing second】second routing message
【routing third】third routing message

分析:

rabbitTemplate.convertAndSend("routingExchange","firstRouting"," first routing message");

消息從生産者指定 firstRouting 路由鍵，找到對應的綁定隊列 routingFirstQueue, 就被 routingFirstQueue 隊列消費了。

主題模式

基于某個主題接收消息

特點:

路由模式發送的消息，是需要指定固定的 routingKey，如果想要針對一類路由。比如：

隻接收以.com 結尾的消息。
www. 開頭的消息。

主題模式就派上場了，路由模式和主題模式類似，路由模式是設定特定的 routingKey 綁定唯一的隊列，而主題模式的是使用通配符比對一個或者多個隊列。

建立交換機和隊列：

@Beanpublic Queue topicFirstQueue() {  
  return new Queue("topicFirstQueue");}
@Bean
public Queue topicSecondQueue() {    
return new Queue("topicSecondQueue");}

@Bean
public Queue topicThirdQueue() {   
 return new Queue("topicThirdQueue");}
@Bean
public TopicExchange topicExchange() {    
return new TopicExchange("topicExchange");}

使用通配符綁定交換機和交換機：

// .com 為結尾 
@Bean
public Binding topicFirstBind() {       return BindingBuilder.bind(topicFirstQueue()).to(topicExchange()).with("*.com");}

 // www.為開頭 
@Bean
public Binding topicSecondBind() {      
return BindingBuilder.bind(topicSecondQueue()).to(topicExchange()).with("www.#");}

通配符有兩種，* 和#,

表示可以比對一個。

表示可以比對多個。

比如：

#.com 表示接收多個以.com 結尾的字段。例如： taobao.com、www.taobao.com、www.jd.com。
*.com 表示接收一個以.com 結尾的字段。例如： taobao.com、jd.com。多個字段是無法比對的，比如 www.taobao.com、cn.taobao.com。
www.#可以比對多個以 www 開頭的字段。例如 www.taobao、www.jd。
www.* 可以比對一個以 www 開頭的字段。例如：www.taobao、www.jd。多個字段是無法比對的, 比如 www.taobao.com、www.jd.com。
生産消息:

@GetMapping("/topic-first-send")
public String topicFirstSend() {    
rabbitTemplate.convertAndSend("topicExchange","www.taobao.com","www.taobao.com");    

rabbitTemplate.convertAndSend("topicExchange","taobao.com","taobao.com");    

rabbitTemplate.convertAndSend("topicExchange","www.jd","www.jd");   
 return "topic ok";
 }

消費消息：

@RabbitListener(queues = "topicFirstQueue")
public void topicFirstListener(String message) {    
System.out.println("【topic first】" + message);}

@RabbitListener(queues = "topicSecondQueue")
public void topicSecondListener(String message) {  
  System.out.println("【topic second】" + message);}

輸出：

【topic second】www.taobao.com
【topic first】taobao.com
【topic second】www.jd

www.#可以比對多個以 www. 開頭的路由鍵，例如 www.taobao.com、www.jd。而 *.com 隻能比對一個以.com 結尾的路由鍵，例如 taobao.com，而無法比對 www.taobao.com。

6. RPC模式

消息有傳回值

特點

PRC 模式和上面的幾種模式唯一不同的點在于，該模式可以收到消費端的傳回值。
生成端接收消費端的傳回值。
消費端添加傳回值：

@RabbitListener(queuesToDeclare =@Queue("rpcQueue"))
public String rpcListener(String message) { 
 System.out.println("【rpc接收消息】" + message);  
 return "rpc 傳回" + message;
 }

生産端發送消息：

@GetMapping("/rpc-send") 
public void rpcSend() {  
Object receive = rabbitTemplate.convertSendAndReceive("rpcQueue","rpc send message");  
System.out.println("【發送消息消息】" + receive);
 }

輸出：

【rpc接收消息】rpc send message
【發送端接收消息】rpc 傳回rpc send message

交換機類型

上面的訂閱釋出模式、路由模式以及主題模式使用到了不同的交換機, 分别是：

直連交換機 Direct

扇形交換機 Fanout

主題交換器 Topic

Direct Exchange（直連）

直連交換機被應用在路由模式下，該交換機需要通過特定的 routingKey 來綁定隊列，交換機隻有接收到了比對的 routingKey 才會将消息轉發到對應的隊列中，否則就不會轉發消息。

路由模式使用直連交換機, 該模式下根據 routingKey 綁定特定的隊列。

Fanout Exchange(扇形)

扇形交換機沒有路由鍵的概念，隻需将隊列綁定在交換機上，發送到交換機上的消息會轉發到交換機是以綁定的隊列裡面，類似廣播，隻要打開收音機都能接收到廣播消息。扇形交換機應用于釋出訂閱模式。

Topic Exchange(主題)

表示可以比對僅一個。

表示可以比對零個或多個。

總結

簡單模式無需建立交換機，比對生産端和消費的 routingKey 即可。
工作模式多個消費端公平競争同一個消息。
釋出訂閱模式一次向多個消費者發送消息。
路由模式根據特定的路由鍵轉發消息。
主題模式根據通配符，比對路由鍵轉發消息。
RPC 模式生産端接收消費端發送的傳回值。

SpringBoot整合RabbitMQ 實作六種工作模式

1. 簡單（simple）模式

二工作模式

釋出訂閱模式

路由模式

主題模式

表示可以比對多個。

6. RPC模式

交換機類型

Fanout Exchange(扇形)

Topic Exchange(主題)

表示可以比對零個或多個。

總結

繼續閱讀

springboot2整合ActiveMQ

SpringBoot——日志檔案設定

微人事學習筆記（持續更新）登入頁面服務端登入接口制作前後端接口對接Home頁面制作前後端分離權限管理思路探讨權限管理：後端接口設計基礎資訊設定頁面

首次阿裡雲部署學生級web服務

rabbitmq連接配接出錯，記錄一下

Spring Boot：靜态資源

RabbitMQ：交換機（fanout exchange）

mysql5.7的sql優化

jeecg-boot的shiro認證過程2. jwtFilter

SpringSecurity整合短信驗證碼認證

基于Springboot+VUE+Mysql的大學生校園疫情防控管理系統

springboot使用 swagger-ui 2.10.5 有關版本更新帶來的問題

rabbitmq 指令行操作

Rabbitmq指令行手動建立隊列

RabbitMQ指令行的基本操作

spring boot @SpringBootApplication @EnableAutoConfiguration 的工作原理

SpringBoot整合RabbitMQ 實作六種工作模式

1. 簡單（simple）模式

二 工作模式

釋出訂閱模式

路由模式

主題模式

表示可以比對多個。

6. RPC模式

交換機類型

Fanout Exchange(扇形)

Topic Exchange(主題)

表示可以比對零個或多個。

總結

繼續閱讀

二工作模式