RabbitMQ 是当前最流行的消息中间件（Message Broker）之一，支持多种消息协议（如 AMQP、MQTT）。
同时它也是一个轻量级的非常易于部署的开源软件，可以运行在当前大多数操作系统及云端环境中，也能够部署在分布式的集群环境里以达到高可用、可伸缩的需求。
此外，RabbitMQ 还为目前主流的编程语言提供了丰富的开发工具。
一、软件安装
可以进入 官方下载界面 阅读针对自己操作系统版本的安装手册，根据需求选择适合的安装方式。
Windows 系统可以直接在该页面中获取二进制安装包（还需要安装 Erlang 环境），Linux 系统也可以根据发行版的不同添加特定的软件镜像源。
我这里是 Ubuntu 19.04，没有特别的需求，所以直接从系统默认的软件镜像源里下载安装，命令如下：$ sudo apt-get install rabbitmq-server
安装完成以后，运行 systemctl status rabbitmq-server 命令查看 RabbitMQ 服务的运行状态：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18$ systemctl status rabbitmq-server
● rabbitmq-server.service - RabbitMQ Messaging Server
Loaded: loaded (/lib/systemd/system/rabbitmq-server.service; enabled; vendor preset: enabled)
Active: active (running) since Fri 2019-07-26 01:03:27 CST; 2min 55s ago
Main PID: 770 (beam.smp)
Status: "Initialized"
Tasks: 85 (limit: 2302)
Memory: 85.8M
CGroup: /system.slice/rabbitmq-server.service
├─ 741 /bin/sh /usr/sbin/rabbitmq-server
├─ 770 /usr/lib/erlang/erts-10.2.4/bin/beam.smp -W w -A 64 -MBas ageffcbf -MHas ageffcbf -MBlmbcs 512 -MHlmbcs 512 -MMmcs 30 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -K true -- -root /usr/lib/erlang -progname erl -- -home /var/lib/rabbitmq -- -pa /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.8/ebin -noshell -noinput -s rabbit boot -sname rabbit@server1 -boot start_sasl -kernel inet_default_connect_options [{nodelay,true}] -sasl errlog_type error -sasl sasl_error_logger false -rabbit lager_log_root "/var/log/rabbitmq" -rabbit lager_default_file "/var/log/rabbitmq/rabbit@server1.log" -rabbit lager_upgrade_file "/var/log/rabbitmq/rabbit@server1_upgrade.log" -rabbit enabled_plugins_file "/etc/rabbitmq/enabled_plugins" -rabbit plugins_dir "/usr/lib/rabbitmq/plugins:/usr/lib/rabbitmq/lib/rabbitmq_server-3.7.8/plugins" -rabbit plugins_expand_dir "/var/lib/rabbitmq/mnesia/rabbit@server1-plugins-expand" -os_mon start_cpu_sup false -os_mon start_disksup false -os_mon start_memsup false -mnesia dir "/var/lib/rabbitmq/mnesia/rabbit@server1" -kernel inet_dist_listen_min 25672 -kernel inet_dist_listen_max 25672
├─1243 erl_child_setup 65536
├─1286 inet_gethost 4
└─1287 inet_gethost 4
7月 26 01:02:44 server1 systemd[1]: Starting RabbitMQ Messaging Server...
7月 26 01:03:27 server1 systemd[1]: rabbitmq-server.service: Supervising process 770 which is not our child. We'll most likely not notice when it exits.
7月 26 01:03:27 server1 systemd[1]: Started RabbitMQ Messaging Server.
Web Admin
RabbitMQ 还提供了可以远程访问的 Web 管理与监控工具，默认以插件的形式安装到系统中，需要使用 rabbitmq-plugins 命令开启。
具体命令如下：$ sudo rabbitmq-plugins enable rabbitmq_management
RabbitMQ 默认创建了一个用户名密码分别为 guest/guest 的用户，只是该用户只允许本地登录。（我这里是远程。。。）
如果需要远程访问 Web 控制台，可以通过 rabbitmqctl 命令创建一个新的管理账户：$ sudo rabbitmqctl add_user <username> <password>
此时新创建的账户仍无法登录，还需要为其分配用户角色以及对 vhost 的管理权限，命令如下：1
2$ sudo rabbitmqctl set_user_tags <username> administrator
$ sudo rabbitmqctl set_permissions -p / <username> ".*" ".*" ".*"
权限设置完毕后，即可用之前指定的用户名密码远程登录 Web 管理系统，界面如下图：
Web 形式的后台界面为管理工作与监控需求提供了便捷的接口，同时大部分管理操作也可直接通过 rabbitmqctl 命令完成，具体可参考该命令的帮助信息：1
2
3
4
5
6
7
8
9
10
11
12
13
14$ sudo rabbitmqctl
Usage:
rabbitmqctl [-n <node>] [-l] [-q] <command> [<command options>]
...
Commands:
add_user <username> <password>
add_vhost <vhost>
authenticate_user <username> <password>
await_online_nodes <count> [-t <timeout>]
cancel_sync_queue [-p <vhost>] queue
change_cluster_node_type <disc|ram>
change_password <username> <password>
...
二、架构解析
RabbitMQ 是一种高性能、稳定、可伸缩（集群部署）的消息中间件，由 Erlang 语言编写。
Erlang 是一种函数式编程语言，专注于分布式、高容错的软件类实时系统等应用场景。它通过轻量级的进程设计以及进程之间的消息通信，提供了一个高层次的不需要共享状态的并发模型。
RabbitMQ 集群通过 Erlang VM 原生的 IPC (inter-process communication) 机制完成跨节点的消息通信。
松耦合架构
对于传统的应用架构，比如一个 Web 应用的登录程序，往往需要对后端的数据库表格进行多项实时的写入操作。而当用户的访问量大增时，此时的表格更新操作很容易成为瓶颈并影响到整体的响应速度。
相对于登录程序直接更新表格数据的紧耦合架构，可以将前端的请求数据推送到基于消息的中间件或者某个中心化的消息队列应用，再通过中间件分发消息到多个消费者（Consumer）应用，由消费者独立、异步地完成最终的数据库更新操作。
基于消息的中间件对于创建数据导向的灵活的应用架构有非常大的优势。RabbitMQ 支持的松耦合设计可以使应用不再受类似于数据库写操作的性能限制。
同时这种架构也非常易于横向扩展，可以在添加作用于相同数据的应用实例时不影响现有的核心功能。
三、消息应用示例代码
下文中将使用 Python 语言及其 RabbitMQ 客户端 Pika 创建 5 个基本的消息应用，结构由简单到复杂，源代码均参考自官网 RabbitMQ Tutorials 。
安装 pika 库：pip install pika
Hello World
该应用的结构示意图如下：
由 P (Producer) 发送一条消息到队列（Queue），再由队列转发消息到 C (Consumer) 。
发送端代码 send.py 如下：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15#!/usr/bin/env python
import pika
# 初始化与 RabbitMQ 服务器的连接
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 队列声明
channel.queue_declare(queue='hello')
# 发送消息
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()
接收端 reveive.py 代码如下：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
# 接收到消息后触发的回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
# 消费者声明与消息监听
channel.basic_consume(
queue='hello', on_message_callback=callback, auto_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
测试
首先运行 4 次发送程序：1
2
3
4
5
6
7
8$ python send.py
[x] Sent 'Hello World'
$ python send.py
[x] Sent 'Hello World'
$ python send.py
[x] Sent 'Hello World'
$ python send.py
[x] Sent 'Hello World'
从 Web 管理界面中可以看到，此时队列中缓存了 4 条消息。
运行接收端程序：1
2
3
4
5
6$ python receive.py
[x] Waiting for messages. To exit press CTRL+C
[x] Received b'Hello World'
[x] Received b'Hello World'
[x] Received b'Hello World'
[x] Received b'Hello World'
发送端连续 4 次发送的消息被接收端收取，队列中缓存的消息被清空。同时接收端保持运行状态等待新的消息被转发给自己。
消息队列一直处于等待生产者发送消息和将收到或缓存的消息转发给消费者的状态。如未有消费者及时接收和处理被转发的消息，则这部分消息缓存在队列中等待进一步操作。
Work Queue
结构示意图：
本例中将创建一个 Work Queue 用来将消耗时间长的任务以轮询的方式分发给多个消费者处理。
生产者源代码 new_task.py ：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(
exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
))
print(" [x] Sent %r" % message)
connection.close()
消费者源代码 worker.py ：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23#!/usr/bin/env python
import pika
import time
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag=method.delivery_tag) # Message acknowledgment
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)
channel.start_consuming()
Message acknowledgment
消费者在处理接收到的任务或消息时有可能会消耗比较多的时间，在此过程中，如消费者端出现软硬件故障，则会出现消息丢失的情况。
RabbitMQ 支持 Message acknowledgment 。即消费者在接收和处理完一个特定的消息后会向 RabbitMQ 返回一个应答（ack），说明该消息可以从队列中移除。
如果消费者在返回应答之前丢失与队列的连接，则 RabbitMQ 判定对应的消息未由消费者完全处理，会将该消息保留在队列中并重新分发给其他在线的消费者。
Message durability
消息应答的机制可以确保即使消费者宕机的情况下任务仍不会丢失。但是当 RabbitMQ 服务本身出现故障时，队列以及队列中缓存的消息仍旧会被清理掉。
为了保证 RabbitMQ 中队列以及消息的持久化，首先需要在生产者和消费者代码中同时声明队列为 durable ：channel.queue_declare(queue='task_queue', durable=True)
此外还需要将生产者代码中的 delivery_mode 属性设置为 2 确保消息的持久化：properties=pika.BasicProperties(delivery_mode=2,)
测试
打开两个命令行终端，分别运行 worker.py 程序：1
2
3# Shell 1
$ python worker.py
[x] Waiting for messages. To exit press CTRL+C
1 | # Shell 2 |
打开另一个终端窗口运行 new_task.py 程序发送 4 条消息：1
2
3
4
5
6
7
8
9# Shell 3
$ python new_task.py First Message
[x] Sent 'First Message'
$ python new_task.py Second Message
[x] Sent 'Second Message'
$ python new_task.py Third Message
[x] Sent 'Third Message'
$ python new_task.py Forth Message
[x] Sent 'Forth Message'
最终两个消费者分别接收到队列分发的两条消息：1
2
3
4
5
6
7# Shell 1
$ python worker.py
[x] Waiting for messages. To exit press CTRL+C
[x] Received b'First Message'
[x] Done
[x] Received b'Third Message'
[x] Done
1 | # Shell 2 |
Fair dispatch
当 RabbitMQ 以轮询的方式（即平均分配）将队列中的消息转发给多个消费者时，如果这些消费者接收到的任务繁重程度差异很大，则会导致某些消费者端任务的积压。
为了避免这种情况发生，可以使用 basic_qos 方法设置 prefetch 的值，如 worker.py 程序中的以下代码：channel.basic_qos(prefetch_count=1) 。
该行代码可以确保同一个消费者在任意时间点最多只接受 1 个任务分配给自己。即如果某个消费者当前有未处理完的消息，则不再接收新的消息直到当前的任务处理完。
Publish/Subscribe
结构示意图：
Exchange
在之前的示例中，用到了消息队列模型中的以下几个组件：
- producer ：生产者，即发送消息的应用
- queue ：队列，即存储消息的缓存
- consumer ：消费者，即接收消息的应用
实际上在 RabbitMQ 的消息模型中，生产者从来不会将消息直接发送到队列中，而是将消息发送给一个名为 exchange 的组件。
exchange 的一端用来接收生产者发送的消息，一端用来将消息推送到队列中。它通过 exchange type 中的定义判断特定的消息是该推送给某个对应的队列，还是将其广播给多个队列，又或者直接丢弃。
RabbitMQ 主要提供了 4 种 exchange 类型：direct、topic、headers 和 fanout。
本例中使用 fanout，即 exchange 会将接收到的消息以广播的形式发送给所有关联的队列，再由队列传递给消费者处理。
源代码（emit_log.py）如下：1
2
3
4
5
6
7
8
9
10
11
12
13
14#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs', exchange_type='fanout')
message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(" [x] Sent %r" % message)
connection.close()
receive_logs.py：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs', exchange_type='fanout')
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='logs', queue=queue_name)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r" % body)
channel.basic_consume(
queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()
receive_logs.py 文件中有一行 result = channel.queue_declare(queue='', exclusive=True) 代码，用来声明一个临时队列（ queue='' 没有指定名称，因此会由 RabbitMQ 设置随机的名称），同时 exclusive=True 设置该队列在消费者断开连接后自行删除。
测试
同时打开两个命令行窗口分别运行 receive_logs.py 文件：1
2
3# Shell 1
$ python receive_logs.py
[*] Waiting for logs. To exit press CTRL+C
1 | # Shell 2 |
再打开第三个终端执行 emit_log.py 命令 4 次：1
2
3
4
5
6
7
8
9# Shell 3
$ python emit_log.py
[x] Sent 'info: Hello World!'
$ python emit_log.py
[x] Sent 'info: Hello World!'
$ python emit_log.py
[x] Sent 'info: Hello World!'
$ python emit_log.py
[x] Sent 'info: Hello World!'
此时之前运行的两个 receive 程序同时收到发送的 4 条消息：1
2
3
4
5
6$ python receive_logs.py
[*] Waiting for logs. To exit press CTRL+C
[x] b'info: Hello World!'
[x] b'info: Hello World!'
[x] b'info: Hello World!'
[x] b'info: Hello World!'
1 | $ python receive_logs.py |
Routing
结构示意图：
与上一个例子中以广播的形式转发消息不同，本例中允许消费者通过队列有选择地订阅生产者发送的部分消息。
Binding 和 Direct exchange
在 RabbitMQ 中，binding 代表 exchange 与队列的对应关系，即队列会根据 binding 的设置对 exchange 转发的消息有选择性地接收。
因此 binding 的最终效果也依赖于 exchange 的类型。比如之前用到的 fanout 类型，由于是广播的形式（转发给所有关联的队列）并不需要选择的动作，则 binding 的值被忽略。
但是对于 direct 类型的 exchange ，则可以通过 binding 对消息进行筛选。在 direct exchange 下，只有当队列的 binding_key 与消息的 routing_key 一致时，队列才会收到 exchange 转发的消息。
emit_log_direct.py：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(
exchange='direct_logs', routing_key=severity, body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()
receive_logs_direct.py：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
sys.exit(1)
for severity in severities:
channel.queue_bind(
exchange='direct_logs', queue=queue_name, routing_key=severity)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(
queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()
测试
首先运行 receive_logs_direct.py 程序并指定参数为 error（即只接收标记为“error”的消息）：1
2
3# Shell 1
$ python receive_logs_direct.py error
[*] Waiting for logs. To exit press CTRL+C
打开另一终端同样运行 receive_logs_direct.py 程序并指定参数为 info warning（即接收标记为 info 或 warning 的消息）：1
2
3# Shell 2
$ python receive_logs_direct.py info warning
[*] Waiting for logs. To exit press CTRL+C
打开第三个终端并运行 emit_log_direct.py 程序发送 4 条日志消息：1
2
3
4
5
6
7
8
9# Shell 3
$ python emit_log_direct.py error "This is an error"
[x] Sent 'error':'This is an error'
$ python emit_log_direct.py info "Hi, I am an info"
[x] Sent 'info':'Hi, I am an info'
$ python emit_log_direct.py warning "Yeah, it's a warning"
[x] Sent 'warning':"Yeah, it's a warning"
$ python emit_log_direct.py error "Hi, it's an error again"
[x] Sent 'error':"Hi, it's an error again"
此时 Shell 1 中只接收到了标记为 error 的消息：1
2
3
4$ python receive_logs_direct.py error
[*] Waiting for logs. To exit press CTRL+C
[x] 'error':b'This is an error'
[x] 'error':b"Hi, it's an error again"
而 Shell 2 中接收到了标记为 info 和 warning 的消息：1
2
3
4$ python receive_logs_direct.py info warning
[*] Waiting for logs. To exit press CTRL+C
[x] 'info':b'Hi, I am an info'
[x] 'warning':b"Yeah, it's a warning"
Topics
结构示意图：
direct 类型的 exchange 虽然可以根据消息的 routing_key 以及队列的 binding_key 有选择性的推送消息到队列，但是并不适合更复杂的场景。
而 topic 类型的 exchange 与 direct 类型逻辑上大致相同，只是 topic 类型的 exchange 并没有一个明确的 routing_key，而是由几个点号（.）分隔的单词（如 lazy.orange.cat）进行定义。
与之对应的 binding_key 也需要遵循同样的形式，只不过 binding_key 额外支持两个特殊含义的字符：
- 星号（
*)可以表示某一个任意的单词 - 井号（
#）可以表示任意 0 个或多个单词
因此对于上图（Topics）中的情形，routing_key 为 quick.orange.rabbit 的消息会被转发给 Q1 和 Q2 队列，quick.orange.fox 则只会转发给 Q1 队列，lazy.orange.male.rabbit 被转发给 Q2 队列。
emit_log_topic：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
routing_key = sys.argv[1] if len(sys.argv) > 2 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(
exchange='topic_logs', routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()
receive_logs_topic.py：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
result = channel.queue_declare('', exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
sys.exit(1)
for binding_key in binding_keys:
channel.queue_bind(
exchange='topic_logs', queue=queue_name, routing_key=binding_key)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(
queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()
测试
先运行接收端程序（Shell 1 和 Shell 2），再运行发送端（Shell 3），效果如下：1
2
3
4
5
6
7# Shell 3
$ python emit_log_topic.py "kern.warning" "A kernel warning message"
[x] Sent 'kern.warning':'A kernel warning message'
$ python emit_log_topic.py "network.critical" "A critical network error"
[x] Sent 'network.critical':'A critical network error'
$ python emit_log_topic.py "kern.critical" "A critical kernel error"
[x] Sent 'kern.critical':'A critical kernel error'
1 | # Shell 1 |
1 | # Shell 2 |