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

Question

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

长版：

Pika 示例都没有演示我的用例。

我有一个 Tornado 服务器，它通过 AMQP（RabbitMQ、Pika）与其他进程/机器通信。这些其他进程的定义不是很明确，但它们在大多数情况下会返回数据（请参阅RabbitMQ 网站上的 RPC 示例）。有时，一个进程可能需要花费极长的时间来处理大量信息，但它不应该完全阻止较小的请求被该进程处理。或者，远程服务器可能因为发送了 Web 请求而阻塞。把它想象成一个 Web 服务器，但是使用 AMQP 而不是 HTTP。

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

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

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，然后推送到主线程可以的公共队列监视器。Tornado 可能会给你这个，但我同意这有点矫枉过正，并且制作这样一个连接包装器不应该那么困难，因为我在不到 100 行 Python 中为其他 I/O 操作做了类似的事情（请参阅线程包装器版本的队列包）。我认为您已经看到了这种可能性，尽管基于您对多个 IOLoops 的讨论。