python - 我如何在长期工作的消费者中提供与 Pika 的并发性？

Question

简短版本:如何防止在远程过程调用情况下阻塞 Pika？

长版:

没有一个 Pika 示例展示了我的用例。

我有一个 Tornado 服务器，它通过 AMQP(RabbitMQ、Pika)与其他进程/机器通信。这些其他进程的定义不是很明确，但在大多数情况下，它们将返回数据(请参阅 RPC example on RabbitMQ's website )。有时，一个进程可能需要花费极长的时间来处理大量信息，但它不应该完全阻止进程处理较小的请求。或者，远程服务器可能因为发出 Web 请求而阻塞。可以将其视为 Web 服务器，但使用 AMQP 而不是 HTTP。

由于 Pika 文档声称它不是线程安全的，因此我无法将连接传递给多个线程(或进程，就此而言)。我想要做的是启动一个新进程，并向 Pika IOLoop 添加一个套接字事件(用于该程序的管道)，就像我可以用 Tornado 做的那样。 Pika IOLoop 与 Tornado IOLoop 有很大不同，它似乎不支持添加多个处理程序；它似乎在一个套接字上使用一个“轮询器”进行操作。

我想避免为这个包要求 Tornado 包，因为我只会使用 IOLoop。这不是不可能的，但我想看看我的其他选择是什么，或者是否可以通过某种方式连接多个 Pika IOLoops/Pollers 来解决我的问题。 RabbitMQ 的文档说，worker 通常可以通过添加更多来“扩展”。我想避免为每个传入的请求创建连接(如果它们传入速度很快)。

Answer 1

根据您的描述，我相信您很遗憾要么需要不同的通信模型，要么需要多个 Pika IOLoops/Pollers/Redundant Connections。

从文档和其他站点听起来，Pika 中的 RPC 始终是一个阻塞语句，无法在线程之间传递。参见 http://www.rabbitmq.com/tutorials/tutorial-six-python.html作者指出，一旦您实际调用 ioloop，Pika 中的 RPC 本质上是阻塞的。

"When in doubt avoid RPC. If you can, you should use an asynchronous pipeline - instead of RPC-like blocking"

如果您想在一个连接完成之前继续在同一个连接上发送多个 RPC 调用，您将需要一个不同的异步模型。在完成之前在同一个连接上进行多次 RPC 调用并不是 RPC 模型的通常实现，尽管这在技术上并不被禁止(http://pic.dhe.ibm.com/infocenter/aix/v6r1/index.jsp?topic=%2Fcom.ibm.aix.progcomm%2Fdoc%2Fprogcomc%2Frpc_mod.htm)。我不认为 Pika 使用此模型运行，尽管它确实通过回调提供了异步支持(我认为这不是您正在寻找的)。

如果您只是想轻松地动态生成新连接，您可以在连接上使用线程或进程包装器，在其他上下文中创建和阻止 RPC，然后推送到公共(public)队列主线程可以监听。 Tornado 可能会给你这个，但我同意这有点矫枉过正，制作这样一个连接包装器应该不会那么困难，因为我已经用不到 100 行 Python 为其他 I/O 操作做了类似的事情(请参阅线程包装器版本的队列包)。我认为您已经根据您对多个 IOLoop 的讨论看到了这种可能性。