AMQP訊息佇列的測試方法

摘要： 在大型網際網路架構中經常會用到訊息佇列（Message Queue）這種中介軟體，在服務端測試時，許多測試同學通過工具對API和資料庫都能熟練地進行測試，一說到訊息佇列的測試就有點不知道怎麼入手了。那麼對於看不見摸不著的訊息佇列，如何進行有效的測試呢？在介紹測試方法之前，我們先來了解一下訊息佇...

在大型網際網路架構中經常會用到訊息佇列（Message Queue）這種中介軟體，在服務端測試時，許多測試同學通過工具對API和資料庫都能熟練地進行測試，一說到訊息佇列的測試就有點不知道怎麼入手了。那麼對於看不見摸不著的訊息佇列，如何進行有效的測試呢？在介紹測試方法之前，我們先來了解一下訊息佇列的原理與機制，這裡以常見的AMQP協議的訊息佇列為例。

1. AMQP訊息佇列簡介

1.1 什麼是訊息佇列

訊息佇列，簡單來說，就是我們通過網路向對方傳送了一封短訊息，短訊息通過運營商網路傳送到接收者，被對方讀取。訊息佇列則是由生產者（訊息的傳送者）通過訊息佇列伺服器向消費者傳送一個訊息，訊息體可以為字串或者更多的資料結構，由消費者在消費端讀取訊息。

1.2 什麼是AMQP

當前各種應用大量使用非同步訊息模型，並隨之產生眾多訊息中介軟體產品及協議，標準的不一致使應用與中介軟體之間的耦合限制產品的選擇，並增加維護成本。 AMQP（Advanced Message Queuing Protocol）是一個提供統一訊息服務的應用層標準協議 ，基於此協議的客戶端與訊息中介軟體可傳遞訊息，並不受客戶端/中介軟體不同產品，不同開發語言等條件的限制。

RabbitMQ是比較常見的一種基於AMQP的訊息服務端。

1.3 AMQP 0.9.1 工作模型

工作過程

1. Publisher將message釋出到exchange（exchange可以看作是一個郵局或者郵件系統，也就是Broker）
2. 將queue註冊到exchange上監聽某種型別的訊息，這個過程稱之為bingding（繫結）
3. exchange將message投遞到queue，這個過程稱之為routing（路由）。
4. AMQP Broker將message傳送給訂閱（subscribed）message的consumer，或者consumer按需將message從對應的queue中取出來。

名詞解釋

• Broker ： 接收和分發訊息的應用，其實就是AMQP伺服器端
• Exchange ： message到達broker的第一站，相當於一個路由器，匹配查詢表中的 routing key ，分發訊息到queue中去。exchange主要有四種類型： direct (點對點)、 topic (主題訂閱) 、 fanout (廣播)和 headers （頭資訊匹配）。
• Queue ： 是一個訊息的載體，訊息最終被送到這裡等待consumer取走。
• Binding ： exchange和queue之間的虛擬連線，binding中可以包含 routing key 。Binding資訊被儲存到exchange中的查詢表中，用於message的分發依據。
• Virtual host ： 當多個不同的使用者使用同一個AMQP服務時，可以劃分出多個vhost，每個使用者在自己的vhost建立exchange／queue等。
• Connection ： publisher／consumer和broker之間的TCP連線。斷開連線的操作只會在client端進行，Broker不會斷開連線，除非出現網路故障或broker服務出現問題。
• Channel ： Channel是在connection內部建立的邏輯連線，如果應用程式支援多執行緒，通常每個thread建立單獨的channel進行通訊，AMQP method包含了channel id幫助客戶端和message broker識別channel，所以channel之間是完全隔離的。

校稿人注

關於channel和多執行緒

如今很少會有單程序單執行緒的應用，大多數情況下生產者、消費者都是多程序多執行緒的，當然每個執行緒都可以建立一個connection，同樣可以滿足連結broker、投遞或消費訊息的工作。但是對作業系統而言，tcp連結是有代價的，而且建立和銷燬tcp連結的代價很昂貴。因此一些具體的訊息佇列實現（例如RabbitMQ）選擇類似IO/">NIO的做法，複用tcp連結。通過在connection之上建立channel邏輯連結，每個執行緒持有自己的channel，實現和broker的通訊，這樣提升了效能，也更便於管理

訊息投遞的幾種方式

1. 點對點（direct）

點對點模式類似於我們發短訊息，由指定的人來接收。

在AMQP中，一個direct型別的exchange基於routing key將投遞訊息到queue中，如果routing key和queue的名字相同，那麼帶有該routing key的訊息會直接被投遞到名字相同的queue中。

2. 廣播（fanout）

顧名思義，廣播模式類似於我們在一個商場裡，突然收到了整個商場的廣播訊息，不管我們願不願意聽，只要在這個商場的人都會聽到。

在AMQP中，一個fanout型別的exchange會忽略routing key，將訊息投遞到所有與之繫結的queue中。

3. 訂閱（topic）

訂閱模式類似於我們刷微博，對於關注的人進行訂閱，當被關注者釋出一條新微博時，所有關注他的人都能夠收到。

在AMQP中，一個topic型別的exchange會對routing key進行模式匹配，將訊息投遞到綁定了對應的routing key的queue中。匹配的時候可以用一些萬用字元，比如“#”表示匹配一個或多個字元。訂閱模式是最常用到的一種訊息投遞方式。

4. 頭資訊匹配（headers）

在AMQP中，一個header型別的exchange會使用訊息的頭部屬性來進行匹配，不再使用routing key。 如果匹配規則比較複雜，需要通過一個hash或者dictionary來匹配的話，可以使用headers這種模式，對頭屬性的值來進行解析和匹配。

校稿人注

就一些特定的AMQP實現而言，例如RabbitMQ，headers模式的路由也是在exchange完成的，但是這種路由模式下exchange的效能會很差，而且這種模式也不實用，所以工程實現上一般都不使用這種模式

2. 訊息佇列的常用測試方法

由於訊息佇列的應用場景主要是圍繞著 生產者 和 消費者 展開的，所以測試思路其實非常簡單，如果被測應用是訊息的生產者，那我們就模擬消費者去接收訊息，驗證發出的訊息內容的正確性。如果被測應用是訊息的消費者，那我們就模擬訊息的生產者去傳送訊息，然後驗證被測應用收到訊息後的處理邏輯。

這裡我們以最常見的RabbitMQ和最常見的Topic Exchange投遞方式為例，介紹一下幾種主要的測試方式。

2.1 被測應用為生產者

2.1.1 手工測試

• 日誌法如果被測應用是生產者，可以讓開發將傳送訊息的內容列印在日誌中，通過檢視日誌的方式進行驗證，這也是比較常用的方法。 但是這種方式如果傳送訊息頻次高、資料量大或者日誌級別設定的不合理的話可能會對應用的效能造成一定影響。

• RabbitMQ管理面板我們需要模擬一個消費者去接收訊息，直接從已有的queue中去取訊息會和其他的消費者產生衝突，所以我們要新建一個測試queue，通過繫結相同的 exchange 和 routing key ，也拿到一份訊息的拷貝。

具體步驟如下：

1. 使用和被測應用相同的vhost賬號登陸RabbitMQ管理面板
2. 新建一個測試queue，命名保證唯一性

   

3. 在測試queue的bindings中，繫結相同的 exchange 和 routing key

   

4. 觸發被測系統傳送訊息，在測試queue中Get Messages來獲取訊息

   

2.1.2 自動化測試

自動化測試的思路其實也是和手工測試一樣，唯一的不同是手工測試時把訊息取出來後是肉眼進行驗證，而自動化測試則需要將訊息落到一個可測的資料載體中，比如資料庫。整體的思路如下圖，是手工測試的一個延展。

這裡我們使用到了python的pika庫（官方文件： ofollow,noindex">pypi.org/project/pik… ） 首先，我們需要安裝pika：

pip install pika
複製程式碼

然後，我們模擬一個阻塞型的消費者來接收訊息，程式碼如下：

# coding:utf-8

__author__ = "小肥羊"

import pika
import json

username = 'username'# 連線RabbitMQ伺服器的使用者名稱
password = 'password'# 連線RabbitMQ伺服器的密碼
host = 'mq_server_host'# 連線RabbitMQ伺服器的地址
port = 'port'# 連線RabbitMQ伺服器的埠號
vhost = 'vhost_name'# vhost名稱

queue_name = 'test_queue'# 新建的測試queue的名稱
exchange_name = 'exchange_name'# exchange名稱
routing_key = 'routing_key_name'# routing key名稱

# 第一步，連線RabbitMQ伺服器
credentials = pika.PlainCredentials(username, password)
connection = pika.BlockingConnection(pika.ConnectionParameters(host, port, vhost, credentials, socket_timeout=120))
# 在連線上建立一個頻道
channel = connection.channel()

# 第二步，為確保佇列存在，再次執行queue_declare建立一個佇列，我們可以多次執行該命令，但是隻有一個佇列會建立
channel.queue_declare(queue=queue_name, durable=True)

# 第三步，為建立的佇列繫結對應的exchange和routing key
channel.queue_bind(queue_name, exchange_name, routing_key)

print ' [*] Waiting for messages. To exit press CTRL+C'


# 第四步，定義一個回撥函式，當獲得訊息時，Pika庫呼叫這個回撥函式來處理訊息，該回調函式將訊息內容列印到螢幕
def callback(ch, method, properties, body):
# 訊息體body轉成json格式
dumped = json.dumps(body, ensure_ascii=False)
pure_json = json.loads(body)
# 將接收到的訊息列印到螢幕
print " [x] Received queue: %r" % (body,)
# 告訴伺服器已經接收到訊息
channel.basic_ack(delivery_tag=method.delivery_tag)


# 第五步，告訴RabbitMQ回撥函式將從queue佇列接收訊息

channel.basic_consume(callback,
queue=queue_name,
no_ack=False)

# 第六步，輸入一個無限迴圈來等待訊息資料並執行回撥函式
channel.start_consuming()
複製程式碼

在callback回撥函式中，只是將訊息內容列印了出來，如果要運用在自動化測試中，我們還需要將訊息內容寫入資料庫中，可以通過sqlalchemy等工具對DB進行寫入操作，這裡就不做詳細介紹了。 另外，由於採用了阻塞型的連線，所以該指令碼最好是部署在測試伺服器上執行，以保證7*24小時的可用性。

2.2 被測應用為消費者

2.2.1 手工測試

如果資料來源依賴於訊息的生產者，那麼我們可以模擬生產者來發送訊息。

在RabbitMQ的管理面板中，允許我們通過 exchange 和繫結的 routing key 來廣播訊息和推送訂閱訊息（ fanout 、 topic 以及 header 模式），也可以直接往queue裡面傳送訊息（ direct 模式），在這裡其實更推薦後者，因為通過前兩者發出的訊息可能有其他的應用系統在消費，可能會對其他應用造成影響，所以建議直接往被測應用監聽的queue裡發訊息。

具體步驟如下：

1. 使用和被測應用相同的vhost賬號登陸管理面板
2. 在queue面板中，找到被測應用監聽的queue
3. 在publish message中的，填入訊息內容併發送

   

4. 驗證被測應用收到訊息後的處理邏輯

2.2.2 自動化測試

自動化模擬生產者要比消費者簡單得多，只需要將訊息傳送到指定的佇列中去，也不需要阻塞式執行指令碼。

實現程式碼如下：

# coding:utf-8

__author__ = '小肥羊'

import pika

username = 'username'# 連線RabbitMQ伺服器的使用者名稱
password = 'password'# 連線RabbitMQ伺服器的密碼
host = 'mq_server_host'# 連線RabbitMQ伺服器的地址
port = 'port'# 連線RabbitMQ伺服器的埠號
vhost = 'vhost_name'# vhost名稱

queue_name = 'queue_name'# 被測系統監聽的佇列名稱

# 第一步，連線RabbitMQ伺服器
credentials = pika.PlainCredentials(username, password)
connection = pika.BlockingConnection(pika.ConnectionParameters(host, port, vhost, credentials, socket_timeout=120))
# 在連線上建立一個頻道
channel = connection.channel()

# 第二步，宣告一個佇列，生產者和消費者都要宣告一個相同的佇列，用來防止萬一某一方掛了，另一方能正常執行
channel.queue_declare(queue=queue_name, durable=True)

# 第三步，傳送訊息，routing_key填的是queue的名稱，這裡exchange填空字串，使用了default exchange
channel.basic_publish(exchange='', routing_key=queue_name, body='message you want to send')

# 第四步，關閉連線
connection.close()

複製程式碼

其中，程式碼中第三步使用了 default exchange ，對此，官方有一個說明：

The default exchange is a direct exchange with no name (empty string) pre-declared by the broker. It has one special property that makes it very useful for simple applications: every queue that is created is automatically bound to it with a routing key which is the same as the queue name.