SpringBoot整合RabbitMQ實現六種工作模式的示例

Posted on 2022-07-28 by WalkonNet

前提概念

生產者

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

隊列

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

消費者

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

SpringBoot整合RabbitMQ基本配置添加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: 192.168.3.19
    port: 5672
    username: admin
    password: 123456

消息生產

生產端發送消息，調用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

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

2. 工作模式

在多個消費者之間分配任務

特點工作模式和簡單模式差不多，隻需要生產端、消費端、隊列。
不同在於一個生產者、一個隊列對應多個消費者，也就是一對多的關系。
在多個消費者之間分配消息（競爭消費者模式），類似輪詢發送消息，每個消息都隻發給一個消費者。

代碼示例

生產消息：

@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 news
second:this is news

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

3. 發佈訂閱模式

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

特點

發佈訂閱類似廣播消息，每個消息可以同時發送給訂閱該消息的消費者，
上圖中的X表示交換機，使用的扇形交換機(fanout),它將發送的消息發送到所有綁定交換機的隊列。

代碼示例

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

@Bean
public FanoutExchange fanoutExchange() {
  return new FanoutExchange("PUBLISH_SUBSCRIBE_EXCHANGE");
}
@Bean
public Queue psFirstQueue() {
  return new Queue("psFirstQueue");
}
@Bean
public 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

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

4. 路由模式

根據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隊列。

生產消息:

@GetMapping("/routing-first")
public String routingFirst() {
    // 使用不同的routingKey 轉發到不同的隊列
    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隊列消費瞭。

5. 主題模式

基於某個主題接收消息

特點

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

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

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

代碼示例

創建交換機和隊列：

@Bean
public 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");
}

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

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

@Bean
public Binding topicSecondBind() {
    // www.為開頭
    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(主題)

主題模式是將路由鍵根據一個主題進行分類，和直連模式不同的是,直連模式綁定特定的路由鍵，而主題模式使用通配符綁定路由鍵,綁定鍵有兩種：

* 表示可以匹配僅一個。
# 表示可以匹配零個或多個。

總結

整合SpringBoot實現RabbitMQ六種工作模式，並詳細講解RabbitMQ六種工作模式：

簡單模式

無需創建交換機，匹配生產端和消費的routingKey即可。

工作模式

多個消費端公平競爭同一個消息。

發佈訂閱模式

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

路由模式

根據特定的路由鍵轉發消息。

主題模式

根據通配符，匹配路由鍵轉發消息。

RPC模式

生產端接收消費端發送的返回值。

源碼示例

https://github.com/jeremylai7/springboot-learning/tree/master/spring-rabbitmq/src/main/java/com/jeremy/pattern

參考文獻

RabbitMQ簡介和六種工作模式

詳解RabbitMQ 的四種交換機

到此這篇關於SpringBoot整合RabbitMQ實現六種工作模式的文章就介紹到這瞭,更多相關SpringBoot整合RabbitMQ內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

SpringBoot整合RabbitMQ實現六種工作模式的示例

目錄

前提概念

生產者

隊列

消費者

SpringBoot整合RabbitMQ基本配置添加maven依賴

1. 簡單（simple）模式

2. 工作模式

生產消息：

3. 發佈訂閱模式

特點

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

4. 路由模式

特點

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

5. 主題模式

特點

創建交換機和隊列：

6. RPC模式

特點

消費端添加返回值：

交換機類型

Direct Exchange（直連）

Fanout Exchange(扇形)

Topic Exchange(主題)

總結

源碼示例

參考文獻

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

前提概念

生產者

隊列

消費者

SpringBoot整合RabbitMQ基本配置添加maven依賴

1. 簡單（simple）模式

2. 工作模式

生產消息：

3. 發佈訂閱模式

特點

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

4. 路由模式

特點

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

5. 主題模式

特點

創建交換機和隊列：

6. RPC模式

特點

消費端添加返回值：

交換機類型

Direct Exchange（直連）

Fanout Exchange(扇形)

Topic Exchange(主題)

總結

源碼示例

參考文獻

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆