eventlet 可以管理 AMQP 连接以及异步传入和传出的消息吗？

Question

实际设计：

对于那些回到这个问题的人，下面有用的答案推动我走向一个运行良好的可行设计。三个见解是关键：

1. Eventlet 是一个非常安全的环境 - 如果两个 greenlet 都尝试从同一个套接字同时尝试 recv() 或都尝试 send()，那么 Eventlet 就会优雅地运行杀死第二个 greenlet 但有异常。这是伟大的，并且意味着如果 amqplib“绿化”不佳，将会导致简单的异常，而不是不可能重现的数据交错错误。
2. amqplib 方法大致分为两类：wait() 在 recv() 内部循环，直到组装 AMQP 消息，而其他方法< code>send() 发回消息，并且不会尝试自己的 recv()。这是非常好的运气，因为 amqplib 作者不知道有人会尝试“绿化”他们的库！这意味着消息发送不仅可以安全地免受 wait() 调用的回调的影响，而且还可以从完全不受 wait() 控制的其他 greenlet 安全地发送消息循环。这些安全方法（可以从任何 greenlet 调用，而不仅仅是从 wait() 回调中调用）是：
   1. basic_ack
   2. basic_consume 与 nowait=True
   3. basic_publish
   4. basic_recover
   5. basic_reject
   6. exchange_declare 与 nowait=True
   7. exchange_delete 与 nowait=True
   8. queue_bind 和 nowait=True
   9. queue_unbind 与 nowait=True
   10. queue_declare 和 nowait=True
   11. queue_delete 与 nowait=True
   12. queue_purge 和 nowait=True
3. 信号量可以用作锁：使用计数 1 初始化信号量，然后使用 acquire() 和 release() 来锁定和解锁。我所有想要写入消息的异步 greenlet 都可以使用这样的锁来避免其单独的 send() 调用交错并破坏 AMQP 协议。

所以我的代码大致如下：

amqp = eventlet.patcher.import_patched('amqplib.client_0_8')

class Processor(object):
    def __init__(self):
        write_lock = eventlet.semaphore.Semaphore(1)

    def listening_greenlet(channel):
        # start this using eventlet.spawn_n()
        # create Connection and self.channel
        self.channel.basic_consume(queue, callback=self.consume)
        while True:
            self.channel.wait()

    def safe_publish(channel, *args, **kw):
        with write_lock:  # yes, Eventlet supports this!
            channel.basic_publish(*args, **kw)     

    def consume(message):
        # Returning immediately frees the wait() loop
        eventlet.spawn_n(self.process, message)

    def process(message):
        # do whatever I want
        # whenever I am done, I can async reply:
        self.safe_publish(...)

享受吧！

原始问题：

想象一下，每分钟有数百条 AMQP 消息到达一个小型 Python Eventlet 应用程序，每个都需要被处理和应答——其中处理的 CPU 开销将是最小的，但可能涉及等待来自其他服务和套接字的应答。

例如，为了允许同时处理 100 条消息，我当然可以启动 100 个与 RabbitMQ 的独立 TCP 连接，并为每个连接配备一个工作人员，以锁步方式接收、处理和应答单个消息。但为了节省 TCP 连接，我更愿意只创建一个 AMQP 连接，允许 RabbitMQ 全速通过管道传输消息，将这些任务交给工作人员，并在每个工作人员完成时发送回答案：

                                       +--------+
                                +------| worker | <-+
                                |      +--------+   |
                                |      +--------+   |
                                | +----| worker | <-+
                                | |    +--------+   |
                                | |    +--------+   |
                                | | +--| worker | <-+
                                | | |  +--------+   |
                                v v v               |
                           +------------+           |
 RabbitMQ <-AMQP-> socket--| dispatcher |-----------+
                           +------------+

：

• 观察 href="http://eventlet.net/doc/modules/queue.html" rel="noreferrer">Eventlet 队列 可以在工作人员可以完成更多工作时优雅地在工作人员之间分配传入的工作。
• RabbitMQ 的流量控制甚至是可能的：我只能在我的工作人员都忙起来之前才确认消息，然后在发送进一步的 ACK 之前等待，直到队列开始变空。
• 工作几乎肯定会无序完成：一个请求可能会很快完成，而另一个较早到达的事件则需要更长的时间；有些请求可能永远不会完成；因此工作人员将以不可预测的异步顺序传回响应。

在看到这篇有吸引力的博客文章后，我一直计划使用 Eventlet 和 py-amqplib 来编写此内容关于如何轻松地将 AMQP 库引入 Eventlet 处理模型：

http://blog.eventlet。 net/2010/02/09/multiple-concurrent-connections-with-py-amqplib-and-eventlet/

我的问题是，阅读了这两个库的文档后，amqplib源代码，以及大部分 Eventlet 源代码，我无法弄清楚如何将拥有 AMQP 连接的 eventlet（博客文章中名为 connect_to_host() 的 eventlet）传授给 当工作人员完成工作并生成答案时醒来。 amqplib 中的 wait() 方法只能通过 AMQP 套接字上的活动来唤醒。虽然感觉我应该能够让工作人员将他们的答案写入队列，并在新传入消息到达时唤醒 connect_to_host() eventlet 或者当工作人员准备好发送答案时，我找不到任何方法让 eventlet 说“当这些事情发生时唤醒我”。

我确实想到，工作人员可以尝试征用 AMQP 连接对象（甚至原始套接字）并通过 TCP 将自己的消息写回；但似乎有必要使用锁来防止传出工作消息相互交错或与主侦听器 eventlet 写入的 ACK 消息交错，而且我也找不到 Eventlet 中可用锁的位置。

所有这一切让我几乎可以肯定，我正试图以某种完全倒退的方式解决这个问题。像这样的问题（让侦听器调度程序和许多工作人员之间安全共享单个连接）是否根本不映射到协程模型，并且需要成熟的异步库？ （在这种情况下：您是否会推荐一个解决此问题的解决方案，以及如何在传入消息和传出工作人员响应之间进行多路复用？今天早些时候，我在尝试 Pika + ioloop 等组合时发现没有干净的解决方案 - 尽管我刚刚看到了另一个库，stormed_amqp，可能比 Pika 做得更好。）或者我真的需要吗回归现实如果我想要可以执行此模型的干净且可维护的代码，请使用实时 Python 线程？我对所有选择持开放态度。

感谢您的任何帮助或想法！我一直认为我已经完全掌握了 Python 并发性，然后我再次发现我没有。 :) 不管怎样，我希望你喜欢上面的 ASCII 艺术。

Answer 1

阅读您的文章并使用 gevent 与 eventlet 类似的库后，我清楚了一些事情，因为我刚刚解决了类似的问题

一般来说，不需要锁定，因为只有一个 eventlet 或 greenlet 同时运行很长一段时间，他们都没有阻止所有事情似乎同时运行..但是你不想在另一个 greenlet 正在发送时沿着套接字发送数据。你是对的，确实需要锁定。

如果我有这样的问题，仅查看文档是不够的......去查看源代码！不管怎样，它是开源的，你可以通过查看其他人的代码学到更多东西。

这里有一些简化的示例代码，可能会帮助您解决问题。

在您的调度程序中有 2 个队列

self.worker_queue = Queue() # queue for messages to workers
self.server_queue = Queue() # queue for messages to ampq server

，工作人员将其结果放入服务器队列中。

发送和接收代码

def send_into_ampq():
    while True:
       message = dispatcher.get_workger_msg()

       try:
          connection.send(self.encode(message))
       except:
           connection.kill()

def read_from_ampq():
    while True:
        message = connection.wait()

        dispatcher.put_ampq_msg(self.decode(message))

连接代码的发送函数中的

self._writelock = Semaphore(1) 
# this is a gevent locking thing. eventlet must have something like this too..
# just counts - 1 for locks and +1 for releases blocks otherwise blocks until 
# 0 agian.... why not google it i though.. and yes im right:
# eventlet.semaphore.Semaphore(value=1)

def send(self, message):
    """
    you need a write lock to prevent more greenlets
    sending more messages when previous sent is not done yet.
    """

    with self._writelock:
        self.socket.sendall(message)