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

1.1 RabbitMQ介紹

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

1.2 RabbitMQ核心概念

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

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

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

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

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

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

Queue：也稱為Message Queue,消息隊(duì)列，保存消息并將它們轉(zhuǎn)發(fā)給消費(fèi)者。（接收端通過(guò)監(jiān)聽(tīng)Queue來(lái)接受消息）。

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

Exchange類(lèi)型：如下四種類(lèi)型：fanout，headers，direct，topic。

圖片

1.3 Exchange交換機(jī)類(lèi)型

• fanout

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

• headers

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

圖片

• direct

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

• topic

1.4 使用案例

準(zhǔn)備環(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類(lèi)型交換機(jī)

• 通過(guò)RabbitMQ控制臺(tái)新建fanout類(lèi)型交換機(jī)：fanout-exchange

圖片

• 新建隊(duì)列：fanout-queue1，fanout-queue2，fanout-queue3

圖片

按照上面的方式分別建立3個(gè)隊(duì)列。

• 將fanout-exchange與3個(gè)隊(duì)列（fanout-queue*）進(jìn)行綁定

圖片

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

• 測(cè)試

發(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("準(zhǔn)備發(fā)送消息：{}", msg);
  rabbitTemplate.convertAndSend("fanout-exchange", "rk.1.2", msg) ;
}

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

圖片

3個(gè)隊(duì)列都收到了消息，說(shuō)明fanout類(lèi)型的交換機(jī)與綁定queue時(shí)routing key是沒(méi)有關(guān)系的，只要綁定了該類(lèi)型的交換機(jī)都會(huì)收到消息。

2、headers類(lèi)型交換機(jī)

• 通過(guò)RabbitMQ控制臺(tái)新建headers類(lèi)型交換機(jī)：headers-exchange

圖片

• 新建隊(duì)列：headers-queue1，headers-queue2

圖片

• 將headers-exchange與2個(gè)隊(duì)列（headers-queue*）進(jìn)行綁定

圖片

• 測(cè)試

發(fā)送接口

@GetMapping("/sendHeaders")
public Object sendHeaders(String msg) {
  ms.sendHeaders(msg) ;
  return "success" ;
}
public void sendHeaders(String msg) {
  logger.info("準(zhǔn)備發(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è)置的參數(shù)完全相同。

圖片

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

圖片

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

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

圖片

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

再次發(fā)送消息：

圖片

兩個(gè)隊(duì)列都收到了消息。

3、direct類(lèi)型交換機(jī)

• 通過(guò)RabbitMQ控制臺(tái)新建direct類(lèi)型交換機(jī)：direct-exchange

圖片

• 新建隊(duì)列：direct-queue1，direct-queue2

圖片

• 將direct-exchange與2個(gè)隊(duì)列（direct-queue*）進(jìn)行綁定

圖片

• 測(cè)試

發(fā)送接口：

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

2個(gè)隊(duì)列都收到了消息。

4、topic類(lèi)型交換機(jī)

• 通過(guò)RabbitMQ控制臺(tái)新建topic類(lèi)型交換機(jī)：topic-exchange

圖片

• 新建隊(duì)列：topic-queue1，topic-queue2

圖片

• 將topic-exchange與2個(gè)隊(duì)列（topic-queue*）進(jìn)行綁定

圖片

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

• 測(cè)試

發(fā)送接口：

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

圖片

兩個(gè)隊(duì)列都收到了消息。

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

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

圖片

只有topic-queue2收到了消息，因?yàn)榻壎〞r(shí)設(shè)置的routing-key 是：

圖片

能夠匹配多個(gè)單詞。

以上就介紹完了rabbitmq的四種交換機(jī)類(lèi)型。