移动测试开发如何使用 RabbitMQ

opentest-oper@360.cn · April 04, 2023 · 4482 hits

一、基本介绍

RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现，最初起源于金融系统，用于在分布式系统中存储转发消息，具备高可用性，易用性、可扩展性的特点。它的运行机制如下图所示：

生产者、消费者不与消息队列 Queue 直连，生产者把消息发送到 Exchange 上，由 Binding 决定发送到哪个队列中并最终由消费者进行消费。

二、安装运行

带有 management 是含有管理界面的，故选择这种进行安装。

使用 docker 安装：

docker run -d --hostname my-rabbit -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq: management

如果在本地的话通过网页http://127.0.0.1:15672 使用默认的 guest/guest 进行登录，至此，rabbitMQ 安装并运行起来。

三、简单的生产消费

生产：

消费：

其中，on_message_callback 的值是回调函数，告诉 channel 接收到数据后，调用这个回调函数。

四、进阶版的生产消费（ack，basic_qos，durable 等）

1.消息确认 ack：一般消费者需要显式通知 rabbitMQ 已经处理完该消息，该任务才能正常被删除。如图：

2.消息持久性 durable：当 RabbitMQ 退出或崩溃时，它会忘记队列和消息，为确保消息不会丢失，我们需要将队列和消息标记为持久：channel.queue_declare(queue='hello', durable=True)，值得注意的是，消息被消费之后收到正常的 ack 该任务消息还是会被删除的。

3.消息分发策略 basic_qos：为了避免出现闲的闲死，忙的忙死的现象，可以使用 basic_qos 方法进行设置，告诉 rabbitMQ 不要同时发送超过一条消息给 worker，直到它已经处理了上条消息并做出了响应：

4.绑定 bindings：告诉交换机如何发送消息给我们的队列。交换机和队列之间的联系我们称之为绑定（binding），使用 queue_bind 方法：
channel.queue_bind(queue=self.taskQueueName,exchange="direct_task",routing_key=self.UUID)

5.临时队列：queue_declare 的参数 exclusive=True 表示当接收端退出时，销毁临时产生的队列，这样就不会占用资源：result = channel.queue_declare(exclusive=True)

五、交换机 Exchange 类型

RabbitMQ 常用的交换器类型有 direct、topic、fanout、headers(目前少用,不做介绍) 四种：
1.Direct 直连交换机：对绑定键（binding key）和路由键（routing key）进行精确匹配，从而确定消息该分发到哪个队列，不符合条件的消息直接过滤掉。

图 1 收到 black 就发送到 Q1，blue 就发送到 Q2，图 2 则是收到 black 发送到两个队列上

Example：发送日志
生产：

消费：

结果展示：（前者生产消息，后者消费显示）

2.Fanout 扇形交换机:将消息发送到所有绑定的队列上，消费者任意消费。除了声明交换机 type 不同之外，其他代码没有太大区别
channel.exchange_declare(exchange='logs',exchange_type='fanout')

3.Topic 主题交换机:模式匹配，将路由键和某个模式匹配，队列绑定模式之后只接收符合该模式的消息。

发送到该交换机的消息不可以携带随意样子的 routing_key,它的 routing_key 必须是由 “.” 隔开的词语列表，如上图发送” beautiful.black.coat” 只将消息传给 Q1,如果是” beautiful.black.shoes “则传给两个队列，” ugly.black.shoes “则只传给 Q2，同时，如果是” ugly.new.orange.shoes“依然可以发送到 Q2

Example:
生产：