python系列之 RabbitMQ -- Publish/Subscribe

矫情吗；* 2022-07-29 10:53 155阅读 0赞

前面的部分我们创建了一个工作队列（work queue）. 设想是每个任务都能分发到一个worker，这一部分我们将会做一些完全不同的事情 -- 我们将会分发一个消息到多个消费方（consumer），这种模式被誉为发布/订阅（publish/subscribe）模式

为了阐明这种模式,我们将要创建一个简单的日志系统，由两部分程序组成 -- 第一部分将要发布日志消息，第二部分接收并打印

在我们的日志系统中每个接收程序（receiver）将接收消息并复制消息内容，这样我们将会运行一个receiver  记录日志到磁盘；与此同时我们运行另一个receiver输入日志到屏幕查看。

本质上，发布日志消息将会广播到所有的receivers

## 交换 (Exchanges) ##

在前面的部分我们从一个队列来回发送并接收消息。现在介绍Rabbit中的完整消息模式。

让我们迅速回顾一下前面的章节的内容：

>>  一个 Producer 是一个发送消息的用户程序

>> 一个 queue 是一个存储消息的缓冲区

>> 一个 Consumer 是一个接收消息的用户程序

RabbitMQ的消息模式的核心思想是生产者(Producer)从不将消息直接发送到一个队列(queue)中，实际上，很多时候生产者甚至不知道一个消息是否要分发到所有队列.

换言之，生产者(producer)只能够发送消息到一个交换区Exchange.. 对exchange发送消息是一个非常容易的事情.  从生产者producer方接收消息，从另一边将消息push到队列中。exchange必须清楚知道接收到的消息要如何处理. 是要将消息发送到一个指定queue？ 是要将消息发送到多个queue？ 还是丢弃？ 这个规则需要通过 exchange type 来定义

![Center][]

这里有几种exchange类型可用： direct  ,  topic ,  headers  和 fanout  。 我们将要对最后一种进行讲解  ---  fanout  。 我们创建一个 这种类型的exchange 并命名为logs：

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

fanout exchange非常简单，你从这个名字中就能猜出来，它将从Producer方收到的消息广播给所有他知道的receiver方。而这正是我们的logger记录所需要的消息。

关于Exchange的几种模式：http://blog.csdn.net/songfreeman/article/details/50953288

exchanges列表

使用rabbitmqclt管理工具显示服务器上的exchanges列表

$ sudo rabbitmqctl list_exchanges
    Listing exchanges ...
    logs      fanout
    amq.direct      direct
    amq.topic       topic
    amq.fanout      fanout
    amq.headers     headers
    ...done.

在这个列表中有一些amq.\*  exchange和默认的exchange，这些都是默认创建的，但是这些未必是你所需要的。

匿名的exchange

在前面部分我们知道空的exchange， 但仍然能够发送消息到队列中，只是因为我们使用的是我们定义的空字符串“”exchange（默认的exchange）

回忆一下我们之前怎么发布一个消息：

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

这个exchange参数就是这个exchange的名字. 空字符串标识默认的或者匿名的exchange：如果存在routing\_key, 消息路由到routing\_key指定的队列中。

现在我们可能够发布消息到我们自己命名的exchange:

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

## 临时队列( Temporary queues ) ##

你应该记得我们之前使用有一个特定名字的队列( hello、task\_queue). 设置队列名对我们来说是至关重要的 --- 我们需要给消费方指定同样的队列名字。 要在生产者和消费者之间共享队列,给队列设置一个名字是非常重要的。

但是这不是我们日志应用的关键,我们希望获取到所有的日志消息，而不是他们的一个子集。我们只对当前活动的消息敢兴趣,对已经发过的旧的消息不关心。为了解决这个问题我们需要做两件事：

首先：

无论什么时候我们连接到Rabbit我们需要一个空的、新的队列。为了实现这个我们可以创建个随机名的队列，或者，更好的 - 让服务端选择一个随机的队列名给我们，我们可以不给queue\_declare方法设置queue参数来实现。

result = channel.queue_declare()

这样, result.method.queue 包含一个随机的队列名, 比如：看起来像  amq.gen-JzTY20BRgKO-HjmUJj0wLg.

其次：

一旦我们断开consumer连接，这个队列名将自动删除。这里有一个标识设置：

result = channel.queue_declare(exclusive=True)

## 绑定（Bindings） ##

![Center 1][]

我们已经创建了一个 fanout exchange 和一个队列queue, 现在我们需要告诉exchange发送消息到我们的queue中, 这个exchange与队列queue之间的关系我们成为 绑定（Bindings）

channel.queue_bind(exchange='logs',
                       queue=result.method.queue)

现在logs exchange 将要发送消息到我们的队列

你可以在Server端通过rabbitmqctl list_bindings命令查看绑定信息

## 汇总(Putting it all together） ##

![Center 2][]

生产者(Producer)程序发出log消息，和前面介绍的章节没什么太大的区别。 主要的改变是我们想要发送消息到我们指定的logs exchange，而不是之前的那种匿名的exchange。 我们之前使用匿名exchange发送消息时Producer需要提供一个 routing\_key，但当我们指定exchange为fanout exchange时，这个值(routing\_key)将忽略。下面是 emit\_log.py 脚本：

import pika
    import sys
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='logs',
                             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()

如你所见, 当建立连接之后我们定义了一个exchange名logs, 由于发布一个消息到一个不存在的exchange是禁止的，所以这一步是必须有的。

发送消息时，如果还没有队列绑定到这个exchange上的话，消息将会丢失。 但这个对我们来说是OK的；如果还没有消费者（consumer) 监听上我们可以安全的放弃这条消息。

receive\_logs.py 脚本 :

import pika
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='logs',
                             type='fanout')
    
    result = channel.queue_declare(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(callback,
                          queue=queue_name,
                          no_ack=True)
    
    channel.start_consuming()

我们已经完成了，如果你想保存日志到一个文件，只需要打开终端执行：

$ python receive_logs.py > logs_from_rabbit.log

如果你想在屏幕上查看输出的日志,新开一个终端并运行：

$ python receive_logs.py

当然，发出日志信息：

$ python emit_log.py

使用 rabbitmqlctl list\_bindings 你能验证代码确实创建了你想要的binding和队列。运行两个 receive\_logs.py 程序你可以看到：

$ sudo rabbitmqctl list_bindings
    Listing bindings ...
    logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []
    logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []
    ...done.

这个结果的解释非常直白： 从 logs  exchange 出来的数据发送服务端自动分配的到两个队列名中，这也是我们预期的。

[Center]: /images/20220728/0a04be6af7b64c4b81e773f44e5edc04.png
[Center 1]: /images/20220728/a708d5093eac4b279c39efe8ac32d824.png
[Center 2]: /images/20220728/c74cbc0cd4ae4883903ddd8b79111fce.png