RabbitMQ 各种通信模式的Python实现

一、RabbitMQ 原理

1、基本原理

RabbitMQ是流行的开源消息队列系统，用erlang语言开发。RabbitMQ是AMQP（高级消息队列协议）的标准实现。支持多种客户端，如：Python、Java、Javascript、C#、C/C++,Go等，支持AJAX，持久化存储。可用于进程之间、分布式系统、异系统之间通信、工作流等。

RabbitMQ支持很多通讯协议，包括AMQP 0-9-1、AMQP 1.0、MQTT和STOMP等。默认使用 AMQP 0-9-1 做为网络层协议。

其支持的网络通讯模型主要有：

• 生产者–消费者模式
• 任务队列模式
• 发布者–订阅者模式
• 路由模式
• RPC模式

所以，如果你的项目包含多个子系统，需要交换的数据有各种类型，有1对1，1对N通信等各种要求，显然成熟的RabbitMQ是1个非常好的选择。如果需要传输大尺寸图像文件，高实时性场景，建议便用ZeroMQ等低层网络库开发消息队列服务器代码更合适。

2、核心组件包括：

• Exchange（交换机）
• Message Queue（消息队列）
• Binding（绑定）

Exchange 交换机类型

• Direct Excnahge直接交换 基于route key 来将消息发送到queue。主要用于单播
• Fanout Exchange 广播交换 不使用route key, 而是一些队列会绑定到Fanout, 新消息会被发送到所有绑定的queue, 适用于广播消息。
• Topic Exchange 主题交换 基于route key 与 匹配pattern ， 将queue绑定到exchange ， 示例用途： 分发与特定地理相关的数据 位置，例如销售点 由多个工作人员完成的后台任务处理， 每个都能够处理特定的任务集 股票价格更新（以及其他类型的财务数据更新） 涉及分类或标记的新闻更新 （例如，仅适用于特定运动或团队）
• Headers exchange 消息头交换 不使用route key, 而是通过message header 来绑定queue 与exchange 。1条queue可以绑定多个header

消息队列 Queue

工作流程
消息队列是FIFO（First In First Out，先进先出）队列，它的作用是：

• 接收消息（from Exchange）
• 保存消息
• 发送消息（to Consumer）

RabbitMQ中Message Queue的基本工作流程是

Queue 的属性

"queues":[{"name":"testQueue","vhost":"/","durable":true,"auto_delete":false,"arguments":{"x-queue-type":"classic"}}]

Binding 绑定

Exchange和Message Queue并没有存储对方的信息，那么Exchange在转发过程中是如何找到正确的Message Queue的呢？这需要借助Binding组件。

Binding中保存着source和destination属性，可以将交换机作为消息源，交换机/消息队列作为转发地址。当交换机路由消息时，会遍历Binding数组，找到source为自身的绑定关系，判断消息属性是否满足routing_key或arguments进行转发。
主要属性

"bindings":[{"source":"amq.headers","vhost":"/","destination":"bigAndBlue","destination_type":"queue","routing_key":"","arguments":{"color":"blue","size":"big","x-match":"all"}}]

RabbitMQ 其它重要概念：

• Broker：简单来说就是消息队列服务器实体。 Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
• Queue：消息队列载体，每个消息都会被投入到一个或多个队列
• Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
• Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
• vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
• producer：消息生产者，就是投递消息的程序。 consumer：消息消费者，就是接受消息的程序。
• channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

二、RabbitMQ 的安装

这里简略介绍Win10下安装 RabbitMQ 3.13.1 的步骤，详细也可参考另1篇介绍安装的文章

1、RabbitMQ安装方式

• Docker 安装方式, 有官方docker image，最方便。
• Linux安装 , Debian或ubuntu
• windows安装，开发环境

当前最新版本 3.13.1, 要求erlang 版本为25.x, 26.x

2、Windows安装步骤

1) 安装 Erlang语言环境

Step-1 从Erlang主页下载26.x 版本。
https://www.erlang.org/downloads

Step-2 下载 windows installer 后安装

step-3 添加环境变量
(1) 新建ERLANG_HOME，指向 erlang安装目录，
(2) 将 %ERLANG_HOME%\bin目录添加至path 系统环境变量。

3、安装 rabbitMQ server.

1） 下载RabbitMQ window installer 安装。
https://www.rabbitmq.com/docs/install-windows

2） 安装后点击安装，系统会自动添加RabbitMQ服务。

3） 按Ctrl+R，输入services, 检查 RabbitMQ 服务是否已启动。

4、基本配置

RabbitMQ 有默认配置。 通常开发环境、单服务器环境下也够用了。
默认配置文件：
windows:

C:\users\username\APPDATA\RabbitMQ\rabbitmq.conf

,
linux 通常为：

/etc/rabbitmq/rabbitmq.conf

RabbitMQ配置较多，如果不熟悉，也可以参考配置文件样例：

• rabbitmq.conf.example

关于配置文件格式，老版本的格已经不支持了。
老版本配置文件使用的格式

%% this is a comment
[{rabbit,[{tcp_listeners,[5673]}]}].

新版本配置文件格式

# this is a comment
listeners.tcp.default =5673

说明： 每1行配置用 parameter = value 定义。 # 开头为注释

5、命令行工具

RabbitMQ提供了一些命令行工具。在安装目录的 sbin/ 子目录下。如 D:\App\rabbitmq\rabbitmq_server-3.13.1\sbin>，

• rabbitmqctl 管理工具
• rabbitmq-diagnostics 健康检查工具
• rabbitmq-plugins 插件管理

使用管理界面来管理rabbitmq
rabbitmq-plugins enable rabbitmq_management, 运行后，默认管理界面的URL: http://localhost:15672/

6）创建vhost 与 用户

安装后，默认用户/密码：guest/guest, 只能从本机访问，本机测试可以使用guest帐号。如果你的python程序与RabbitMQ不在1台机器上，则需要用上节方法登录rabbitmq 管理界面，创建vhost, 以及用户帐号，添加用户权限( 由于比较简单就不多说了）

也可以用命令行创建
进入安装目录的 sbin\ 目录下，运行

rabbitmqctl add_user myuser mypassword
rabbitmqctl add_vhost myvhost
rabbitmqctl set_user_tags myuser mytag
rabbitmqctl set_permissions -p myvhost myuser ".*"".*"".*"

将myuser, mypassword, myvhost 改成你需要的即可。

赋予其administrator角色：
rabbitmqctl set_user_tags user_admin administrator

查看已有用户
rabbitmqctl list_users

三、RabbitMQ 各类通信模式的实现

1、安装 RabbitMQ 客户端连接工具

Step-1 安装RadditMQ 客户端
python -m pip install pika --upgrade

Step-2: 在python代码中导入pika
import pika

Step-3: 创建连接

# 使用默认本地帐号与密码， 否则使用前1章创建的用户名
credentials = pika.PlainCredentials(username="guest", password="guest")# 创建连接
connect = pika.BlockingConnection(pika.ConnectionParameters(host='localhost', port=5672, virtual_host='/', credentials=credentials))# 创建1wh channel用于具体网络操作
channel = connect.channel()

2、基本模式：生产者–消费者模式

本例功能需求； 生产者将消息发往Queue, 消费者从queue接收消息

1） 生产者代码实现 producer.py

import pika 

# 首先建立至RabbitMQ 服务器的连接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()# 创建1个 queue
channel.queue_declare(queue='hello')#Rabbit发送消息，须经过exchage, 本例 使用默认exchange， 使用routing_key=’hello’发送消息。
channel.basic_publish(exchange='',
                      routing_key='hello',# hello为前面创建的queue名字
                      body='Hello World!')print(" [x] Sent 'Hello World!'")#发送完成后，即可关闭连接 
connection.close()

2）消费者接收消息代码实现

import pikia 
#建立至rabiitMQ服务器的连接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()# 声明接受队列queue
channel.queue_declare(queue='hello')#定义callback defcallback(ch, method, properties, body):print(" [x] Received %r"% body)# 配置consume 参数，指定queue, 回调函数，auto_ack等。 
channel.basic_consume(queue='hello',
                      auto_ack=True,
                      on_message_callback=callback)# 等待数据，收到后自动执行callback 
channel.start_consuming()

3) 测试：

1）打开第1个终端 ，运行客户端 python receive.py
2）打开第2个终端 ，运行Producer端， python send.py

consumer端应该显示

# => [*] Waiting for messages. To exit press CTRL+C
# => [x] Received 'Hello World!'

3、任务队列模式的实现

任务队列 Work Queue， 也称Task Queue, 主要用于发布耗时任务.

功能需求：
(1)Producer将任务及数据封装在1 个message中,发送给work queue,
(2)Worker 从队列中读取消息。几个worker同时工作，则速度大大提高。
(3) Work Queue中的1条消息，RabbitMQ Server只发给1个worker, 发送完成后删除。

Producer.py , 创建发布task message.

import sys
message =' '.join(sys.argv[1:])or"Hello World!"
channel.basic_publish(exchange='',
routing_key='hello',
body=message)print(f" [x] Sent {message}")

Worker.py， 处理任务的工作放在callback 函数。

defcallback(ch, method, properties, body):print(f" [x] Received {body.decode()}")
    time.sleep(body.count(b'.'))print(" [x] Done")
ch.basic_ack(delivery_tag = method.delivery_tag)(发送ack)

测试：
启动2个worker.py ,用1个producer发布task消息

消息持久化配置
当rabbitmq server宕机，任务消息会丢失，如果需要保持queue不丢失

Worker端:

channel.queue_declare(queue='hello', durable=True)

Producer端

channel.basic_publish(exchange='',
                      routing_key="task_queue",
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = pika.DeliveryMode.Persistent
                      ))

Pair Dispatch 根据ack分派消息
为避免worker负荷不均,使用pair dispatch 方式： Server只有收到Worker上1条消息的ack ，才发送1条新消息。 worker设置 prefetch_count参数=1
channel.basic_qos(prefetch_count=1)

完整代码
Producer.py

#!/usr/bin/env pythonimport 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=pika.DeliveryMode.Persistent
    ))print(f" [x] Sent {message}")
connection.close()

Worker.py

#!/usr/bin/env pythonimport 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')defcallback(ch, method, properties, body):print(f" [x] Received {body.decode()}")
    time.sleep(body.count(b'.'))print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)

channel.start_consuming()

4、Publish-Subscribe 发布订阅模式的实现

Pub-Sub中，producer发布1条消息，这条消息可以发送给多个consumer.

功能需求： 构建1个log system, 1个emit 发送log, 多个 receiver 接收log并打印。

本例 exchange 使用fanout 类型，使用默认queue, 每条消息都会广播给所有consumer,

Producer端

channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')

发布消息

channel.basic_publish(exchange='logs',  
                      routing_key='',
                      body=message)

完整代码： publish.py

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()#设置exchange参数
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(f" [x] Sent {message}")
connection.close()

Consumer端

先创建exchange对象，申明默认queue, 绑定exchage 与queue.
channel.queue_bind(exchange=‘logs’, queue=result.method.queue)

Binding 关系，可以理解为，这个queue接收从该exchange发送的所有消息。也可以添加route-key参数。

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()#使用与Producer相同的 exchange, 
channel.exchange_declare(exchange='logs', exchange_type='fanout')# 使用默认queue, 绑定至exchange
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue   #系统命名默认queue
channel.queue_bind(exchange='logs', queue=queue_name)print(' [*] Waiting for logs. To exit press CTRL+C')defcallback(ch, method, properties, body):print(f" [x] {body}")

channel.basic_consume(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

5、路由模式的实现

路由 Routing模式使用exchage类型为

direct

, 使用route-key将exchage与队列绑定。

如上图，队列 Q1 与 只包含 orange的消息， Q2 接受包含 black 或 green的消息。

也可以将1个route-key绑定到多个queue.

发布方要点（ pub.py)

创建exchange，本例使用默认queue.

channel.exchange_declare(exchange='direct_logs',
exchange_type='direct')

发送消息

channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,    
                      body=message)

订阅方的要点： receiver.py
订阅方一侧将exchange direct_logs与 默认queue绑定

result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

for severity in severities:
   channel.queue_bind(exchange='direct_logs',
      queue=queue_name,
      routing_key=severity)

完整代码
Receiver.py

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:]ifnot 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')defcallback(ch, method, properties, body):print(f" [x] {method.routing_key}:{body}")

channel.basic_consume(
queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

Pub.py

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]iflen(sys.argv)>1else'info'
message =' '.join(sys.argv[2:])or'Hello World!'
channel.basic_publish(
exchange='direct_logs', routing_key=severity, body=message)print(f" [x] Sent {severity}:{message}")
connection.close()

测试
启动多个receiver， 分别接受不同

worker-1:  python receiver.py error warning 
worker-2:  python receiver.py warning 

用pub.py 发消息

Python sub.py warning “a warning for test”     # worker-1, worker-2都会收到
Python sub.py error “a error for test”         # only worker-1 收到

6、主题模式Topics

主题网络模式使用 topic exchage, 可以用于更复杂的场景。
主题交换的 route-key 使用替换掩码

• * 表示 1个词
• # 表示 0或多个word.

Topic的route-key 建议格式：

<category>.<colour>.<species>

, 每1级主题之间用

.

点号分隔。

如 ：

• *.orange.* 可以匹配 Camelia.orange.Aprilblush
• lazy.# 可以匹配到 lazy.pig.black
• # 表示该queue可以接收所有消息，相当于fanout
• 不含*与# 的 route-key ，与 direct exchange作用相同。

本例 ，我们还是以日志系统为例 ，接收者可以带掩码的route-key 来更灵活地接收自己所需要的消息。

完整实现代码：
Pub.py

import pika
import sys

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

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')#第1个命令行参数为routing-key
routing_key = sys.argv[1]iflen(sys.argv)>2else'anonymous.info'#第2个命令行参数为消息内容
message =' '.join(sys.argv[2:])or'Hello World!'
channel.basic_publish(
    exchange='topic_logs', routing_key=routing_key, body=message)print(f" [x] Sent {routing_key}:{message}")
connection.close()

Receiver.py

#!/usr/bin/env pythonimport 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-key, 可输入多个
binding_keys = sys.argv[1:]ifnot 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')defcallback(ch, method, properties, body):print(f" [x] {method.routing_key}:{body}")

channel.basic_consume(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()

测试
启动消息者 receiver.py

python receive_logs_topic.py "kern.*"

或者，你想接收 critical 日志

python receive_logs_topic.py "*.critical"

可以建立多个绑定

python receive_logs_topic.py "kern.*""*.critical"

启动发布者，发布消息

python pub.py "kern.critical""A critical kernel error"

可以看到消费者收到了相关消息。

7、RPC 调用

在rpc场景中，Server暴露1个接口， client 在调用时，将调用请求做为消息发布至 1 queue, 同时指定reply_to 队列，Server将响应发送到reply_to 队列
Server 在开始是做为消息的接受者，发送响应时做为消息发送者。

client.py 代码

import pika
import uuid

classFibonacciRpcClient(object):def__init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host='localhost'))

        self.channel = self.connection.channel()

        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True)

        self.response =None
        self.corr_id =Nonedefon_response(self, ch, method, props, body):if self.corr_id == props.correlation_id:
            self.response = body

    defcall(self, n):
        self.response =None
        self.corr_id =str(uuid.uuid4())# 产生1个 uuidprint("发送rpc请求")
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.corr_id,),
            body=str(n))while self.response isNone:
            self.connection.process_data_events(time_limit=None)returnint(self.response)

fibonacci_rpc = FibonacciRpcClient()print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)print(f" [.] Got {response}")

提供 rpc 函数的服务端代码 server.py

import pika

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

channel = connection.channel()

channel.queue_declare(queue='rpc_queue')deffib(n):if n ==0:return0elif n ==1:return1else:return fib(n -1)+ fib(n -2)defon_request(ch, method, props, body):
    n =int(body)print(f" [.] 收到请求，调用 fib({n})")
    response = fib(n)

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(
                         correlation_id=props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)print(" [x] Awaiting RPC requests")
channel.start_consuming()

测试：
先启动服务端 python server.py
再启动客户端 python client.py