AMQP消息队列的测试方法

在大型互联网架构中经常会用到消息队列（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）选择类似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库（官方文档： 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.