環(huán)境：Spring Boot2.3.10 + RabbitMQ 3.8.12 + Erlang 23.2.5

1.1 RabbitMQ介紹

RabbitMQ是實現(xiàn)了高級消息隊列協(xié)議（AMQP）的開源消息代理軟件（亦稱面向消息的中間件）。RabbitMQ服務(wù)器是用Erlang語言編寫的。所有主要的編程語言均有與代理接口通訊的客戶端庫。

1.2 RabbitMQ核心概念

Server：又稱Broker, 接受客戶端的連接，實現(xiàn)AMQP實體服務(wù)，這里指RabbitMQ 服務(wù)器

Connection：連接，應(yīng)用程序與Broker的網(wǎng)絡(luò)連接。

Channel：通信通道，幾乎所有的操作都在 Channel 中進行，Channel是進行消息讀寫的通道。客戶端可建立多個Channel:，每個Channel代表一個會話任務(wù)。（一個Socket連接可以創(chuàng)建多個Channel）

Virtual Host：虛擬主機，用于邏輯隔離，是最上層的消息路由。一個 Virtual Host 里面可以有若干個 Exchange（交換機）和 Queue（隊列）。

Binding：Exchange 和 Queue 之間的虛擬綁定關(guān)系。

Routing key：一 個路由規(guī)則，虛擬機可用它來確定如何路由一個特定消息，即交換機綁定到 Queue 的鍵。支持模式匹配，* 和#，*用來匹配一個單詞，#用來匹配多個單詞（可以是0個）這里的通配符只有在exchange的類型為topic才有效。

Queue：也稱為Message Queue,消息隊列，保存消息并將它們轉(zhuǎn)發(fā)給消費者。（接收端通過監(jiān)聽Queue來接受消息）。

Message：消息，服務(wù)器和應(yīng)用程序之間傳送的數(shù)據(jù)，由 Properties 和 Body 組成。Properties 可以對消息進行修飾，比如消息的優(yōu)先級、延遲，過期時間，消息大小等高級特性;，Body 則就 是消息體內(nèi)容。

Exchange類型：如下四種類型：fanout，headers，direct，topic。

圖片

1.3 Exchange交換機類型

• fanout

發(fā)布/訂閱，消費發(fā)送到該種類型的交換機后會將消息發(fā)送到所有與之綁定的Queue中，不管你有沒有設(shè)置routing key。

• headers

該類型用得極少，主要就是根據(jù)設(shè)置的header來進行匹配。如下紅框中你所設(shè)置的arguments：

圖片

• direct

在exchange與queue進行綁定時需要設(shè)置routing key，只有發(fā)送消息的端設(shè)置的routing key與這里綁定的routingkey完全相同才會將消息發(fā)到隊列中。

• topic

1.4 使用案例

準備環(huán)境：

依賴

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>

配置文件

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: 123456
    virtual-host: /
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        concurrency: 5
        maxConcurrency: 10
        prefetch: 5
        acknowledgeMode: MANUAL
        retry:
          enabled: true
          initialInterval: 3000
          maxAttempts: 3
        defaultRequeueRejected: false

1、fanout類型交換機

• 通過RabbitMQ控制臺新建fanout類型交換機：fanout-exchange

圖片

• 新建隊列：fanout-queue1，fanout-queue2，fanout-queue3

圖片

按照上面的方式分別建立3個隊列。

• 將fanout-exchange與3個隊列（fanout-queue*）進行綁定

圖片

將3個隊列分別于fanout-exchange進行綁定，這里在綁定時分別設(shè)定了不同的路由key，就是為了驗證fanout類型的交換機是否與routing key有關(guān)系。

• 測試

發(fā)送接口：

@RestController
@RequestMapping("/messages")
public class MessageController {
  @Resource
  private MessageSend ms ;
  @GetMapping("/sendFanout")
  public Object send(String msg) {
    ms.send(msg) ;
    return "success" ;
}
}
@Resource
private RabbitTemplate rabbitTemplate ;


public void send(String msg) {
  logger.info("準備發(fā)送消息：{}", msg);
  rabbitTemplate.convertAndSend("fanout-exchange", "rk.1.2", msg) ;
}

這里故意設(shè)置了路由key 就是為了看fanout類型的交換機是否與routing key 有關(guān)系。

圖片

3個隊列都收到了消息，說明fanout類型的交換機與綁定queue時routing key是沒有關(guān)系的，只要綁定了該類型的交換機都會收到消息。

2、headers類型交換機

• 通過RabbitMQ控制臺新建headers類型交換機：headers-exchange

圖片

• 新建隊列：headers-queue1，headers-queue2

圖片

• 將headers-exchange與2個隊列（headers-queue*）進行綁定

圖片

• 測試

發(fā)送接口

@GetMapping("/sendHeaders")
public Object sendHeaders(String msg) {
  ms.sendHeaders(msg) ;
  return "success" ;
}
public void sendHeaders(String msg) {
  logger.info("準備發(fā)送消息：{}", msg);
  Message message = MessageBuilder.withBody(msg.getBytes()).setHeader("a", "b").setHeader("pack", "xg").build() ;
  rabbitTemplate.send("headers-exchange", "rk.1.2", message) ;
}

注意：這里設(shè)置了Header信息與headers-exchange和headers-queue1綁定時設(shè)置的參數(shù)完全相同。

圖片

headers-queue2就沒有收到消息。如果在進行綁定時沒有設(shè)置x-match參數(shù)默認是完全要匹配綁定時設(shè)定的參數(shù)，也就是x-match:all。如下，如果不配置x-match也就相當(dāng)于：

圖片

x-match 還有一個取值就是any，只要有任意的一個匹配上那就成功。

修改headers-queue1，headers-queue2綁定的參數(shù)，如下：

圖片

兩個參數(shù)中都加入了x-match:any 只要有任意一個參數(shù)匹配即可。

再次發(fā)送消息：

圖片

兩個隊列都收到了消息。

3、direct類型交換機

• 通過RabbitMQ控制臺新建direct類型交換機：direct-exchange

圖片

• 新建隊列：direct-queue1，direct-queue2

圖片

• 將direct-exchange與2個隊列（direct-queue*）進行綁定

圖片

• 測試

發(fā)送接口：

@GetMapping("/sendDirect")
public Object sendDirect(String msg) {
  ms.sendDirect(msg) ;
  return "success" ;
}
public void sendDirect(String msg) {
  logger.info("準備發(fā)送消息：{}", msg);
  rabbitTemplate.convertAndSend("direct-exchange", "de.m", msg) ;
}

2個隊列都收到了消息。

4、topic類型交換機

• 通過RabbitMQ控制臺新建topic類型交換機：topic-exchange

圖片

• 新建隊列：topic-queue1，topic-queue2

圖片

• 將topic-exchange與2個隊列（topic-queue*）進行綁定

圖片

*: 只能匹配一個單詞；#：能匹配多個單詞（也可以是0個）。

• 測試

發(fā)送接口：

@GetMapping("/sendTopic")
public Object sendTopic(String msg) {
  ms.sendTopic(msg) ;
  return "success" ;
}
public void sendTopic(String msg) {
  logger.info("準備發(fā)送消息：{}", msg);
  rabbitTemplate.convertAndSend("topic-exchange", "te.1.ok", msg) ;
}

圖片

兩個隊列都收到了消息。

修改發(fā)送消息的routing key

rabbitTemplate.convertAndSend("topic-exchange", "te.2.ok", msg) ;

圖片

只有topic-queue2收到了消息，因為綁定時設(shè)置的routing-key 是：

圖片

能夠匹配多個單詞。

以上就介紹完了rabbitmq的四種交換機類型。