通過實例理解 RabbitMQ 的基本概念

先說下自己開發的實例。最近在使用 Spring Cloud Config 做分散式配置中心（基於 SVN/Git），當所有服務啟動後，SVN/Git 中的配置文件更改後，客戶端服務讀取的還是舊的配置，並不能實時讀取（配置信息會緩存在客戶端），Spring Boot 提供了一種方式進行更新（通過 ...

先說下自己開發的實例。

最近在使用 Spring Cloud Config 做分散式配置中心（基於 SVN/Git），當所有服務啟動後，SVN/Git 中的配置文件更改後，客戶端服務讀取的還是舊的配置，並不能實時讀取（配置信息會緩存在客戶端），Spring Boot 提供了一種方式進行更新（通過spring-boot-starter-actuator監控模塊），然後 Post 訪問客戶端服務的/refresh介面（也可以命令執行curl -X POST http://worker2:8115/refresh），這樣客戶端會重新從配置中心獲取新的配置信息，請求命令可以寫在 Git 的 Webhooks 腳本中（修改提交 Push 後執行）。

如果客戶端服務比較少的話，這樣的解決方式沒問題，如果客戶端服務多的話，執行的請求腳本就會非常多，而且單個服務的解決方式，也不利於後期的維護（點對點的方式），那該怎麼解決上面的問題呢？答案就是通過 Spring Cloud Bus。

Spring Cloud Bus 翻譯為消息匯流排，使用輕量級的消息代理來構建一個共用的消息主題讓系統中所有微服務實例都能連接上來，由於該主題中產生的消息會被所有實例監聽和消費，所以我們稱它為消息匯流排。在匯流排上的各個實例都可以方便地廣播一些需要讓其他連接在該主題上的實例都知道的消息，例如配置信息的變更或者其他一些管理操作等。

架構示意圖（引用來源）：

下麵，我們需要利用 Spring Cloud Bus 來改造 Spring Cloud Config 的服務端和客戶端，其實非常簡單。

添加下麵的依賴：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

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

然後在bootstrap.yml中添加下麵配置：

spring:
  rabbitmq:
    host: manager1
    port: 5672
    username: admin
    password: admin123
management:
  security:
    enabled: false

上面的配置信息都是新增的，並且都需要配置在服務端和客戶端，通過上面的示例圖可以看到，配置信息更新後請求的是服務端，那麼客戶端我們就不需要配置management.security.enabled（也不需要配置spring-boot-starter-actuator監控模塊）。

服務端和客戶端的任何 Java 代碼都不需要編寫，重新啟動服務，當配置信息更新後，通過 Git 的 Webhooks 執行請求腳本：curl -X POST http://manager1:port/bus/refresh，服務端接受到請求之後，會通過 Spring Cloud Bus 通知所有的客戶端（通過 RabbitMQ），重新從配置中心獲取配置信息，達到實時更新配置的目的。

上面的實例描述就到這裡。

RabbitMQ 的基本概念

RabbitMQ，是一個使用 erlang 編寫的 AMQP（高級消息隊列協議）的服務實現，簡單來說，就是一個功能強大的消息隊列服務。

RabbitMQ 最基本模型：

RabbitMQ 的基本概念：

Producer：消息生產者。
Consumer：消息消費者。
Connection（連接）：Producer 和 Consumer 通過TCP 連接到 RabbitMQ Server。
Channel（通道）：基於 Connection 創建，數據流動都是在 Channel 中進行。
Exchange（交換器）：生產者將消息發送到 Exchange（交換器），由 Exchange 將消息路由到一個或多個 Queue 中（或者丟棄）；Exchange 並不存儲消息；Exchange Types 常用有 Fanout、Direct、Topic 三種類型，每種類型對應不同的路由規則。
Queue（隊列）：是 RabbitMQ 的內部對象，用於存儲消息；消息消費者就是通過訂閱隊列來獲取消息的，RabbitMQ 中的消息都只能存儲在 Queue 中，生產者生產消息並最終投遞到 Queue 中，消費者可以從 Queue 中獲取消息並消費；多個消費者可以訂閱同一個 Queue，這時 Queue 中的消息會被平均分攤給多個消費者進行處理，而不是每個消費者都收到所有的消息並處理。
Binding（綁定）：是 Exchange（交換器）將消息路由給 Queue 所需遵循的規則。
Routing Key（路由鍵）：消息發送給 Exchange（交換器）時，消息將擁有一個路由鍵（預設為空）， Exchange（交換器）根據這個路由鍵將消息發送到匹配的隊列中。
Binding Key（綁定鍵）：指定當前 Exchange（交換器）下，什麼樣的 Routing Key（路由鍵）會被下派到當前綁定的 Queue 中。

另外，再說下 Exchange Types（交換器類型）的三種常用類型：

Direct：完全匹配，消息路由到那些 Routing Key 與 Binding Key 完全匹配的 Queue 中。比如 Routing Key 為cleint-key，只會轉發cleint-key，不會轉發cleint-key.1，也不會轉發cleint-key.1.2。
Topic：模式匹配，Exchange 會把消息發送到一個或者多個滿足通配符規則的 routing-key 的 Queue。其中*表號匹配一個 word，#匹配多個 word 和路徑，路徑之間通過.隔開。如滿足a.*.c的 routing-key 有a.hello.c；滿足#.hello的 routing-key 有a.b.c.helo。
Fanout：忽略匹配，把所有發送到該 Exchange 的消息路由到所有與它綁定的Queue 中。

下麵通過一段代碼，理解一下消息發佈的流程（代碼引用）：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='first', type='topic')
channel.queue_declare(queue='A')
channel.queue_declare(queue='B')

channel.queue_bind(exchange='first', queue='A', routing_key='a.*.*')
channel.queue_bind(exchange='first', queue='B', routing_key='a.#')

channel.basic_publish(exchange='first',
                      routing_key='a',
                      body='Hello World!')

channel.basic_publish(exchange='first',
                      routing_key='a.b.c',
                      body='Hello World!')

大致步驟：

先獲取一個 Connection（連接）。
從 Connection（連接）上獲取一個 Channel（通道）。
聲明一個 Exchange（交換器），只會創建一次。
聲明兩個 Queue，只會創建一次。
把 Queue 綁定到 Exchange（交換器）上.
向指定的 Exchange（交換器）發送一條消息.

因為基於 Exchage Topic 模式，在上面發出的兩條消息當中，消息a只會被a.#匹配到，而a.b.c會被兩個都匹配到。所以，最終的結果會是隊列 A 中有一條消息，隊列 B 中有兩條消息。

從隊列取出消息代碼：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='A')

def callback(ch, method, properties, body):
    print body

channel.basic_consume(callback, queue='A', no_ack=True)
channel.start_consuming()

服務消費者取出消息，需要重新創建 Connection（連接）和 Exchange（交換器），但 Queue 並不會創建，只需要從 Channel 中獲取對應的 Queue 消息即可。

通過實例理解 RabbitMQ 基本概念

上面實例服務部署的情況是：三台管理伺服器（config-server-git/config-server-svn）和一臺工作伺服器（config-client-git/config-server-svn），因為做了集群，服務的具體情況：

config-server-git：3 個服務。
config-server-svn：3 個服務。
config-client-git：1 個服務。
config-client-svn：1 個服務。

所以，總的部署服務有 8 個。

我們通過 RabbitMQ Server 管理界面中的內容，說下 Connection（連接）、Channel（通道）、Exchange（交換器）和 Queue（隊列）的具體使用情況（根據數量理解）。

1. Connection（連接）

為什麼 Connection（連接）數量為 16 個？因為部署的 8 個服務，各自發佈和接受消息（即作為小心發佈者，也作為消息接受者），計算公式：16 = 8 * 2。

2. Channel（通道）

為什麼 Channel（通道）數量為 16 個？因為 Connection（連接）數量為 16 個，Channel（通道）是在 Connection（連接）基礎上創建的。

3. Exchange（交換器）

為什麼 Exchange（交換器）數量為 1 個？因為都是使用的同一個 Exchange（交換器），名字為springCloudBus，Exchange Type 為topic，Routing Key 為#。

4. Queue（隊列）

為什麼 Queue（隊列）數量為 8 個？因為部署的 8 個服務，各自發佈和接受的 Queue 是同一個，一個服務對應一個 Queue。

參考資料：

RabbitMQ 消息隊列 - RabbitMQ 消息隊列架構與基本概念（推薦）
RabbitMQ 使用參考（推薦）
RabbitMq（一）走進 RabbitMq
RabbitMQ 基本概念和使用（推薦）
分散式消息隊列 RabbitMQ 之一：基本概念理解
RabbitMQ 基礎概念詳細介紹
RabbitMQ 基本概念