使用 Tornado 设计异步请求和阻塞处理答案

【问题标题】：Design of asynchronous request and blocking processing using Tornado使用 Tornado 设计异步请求和阻塞处理
【发布时间】：2016-01-12 13:33:48
【问题描述】：

我正在尝试实现一个 Python 应用程序，该应用程序使用基于 Tornado 的 client 使用 NATS 来接收和发送消息。收到消息后，必须调用阻塞函数，我试图在单独的线程上实现该函数，以允许消息的接收和发布将消息放入 Tornado 队列，以便稍后处理阻塞函数。

我对 Tornado（以及 python 多线程）非常陌生，但是在阅读了几次 Tornado 文档和其他资源之后，我已经能够提供一个工作版本的代码，如下所示：

import tornado.gen
import tornado.ioloop
from tornado.queues import Queue
from concurrent.futures import ThreadPoolExecutor
from nats.io.client import Client as NATS

messageQueue = Queue()
nc = NATS()
@tornado.gen.coroutine
def consumer():
    def processMessage(currentMessage):
        # process the message ...

    while True:
        currentMessage = yield messageQueue.get()
        try:
            # execute the call in a separate thread to prevent blocking the queue
            EXECUTOR.submit(processMessage, currentMessage)
        finally:
            messageQueue.task_done()

@tornado.gen.coroutine
def producer():
    @tornado.gen.coroutine
    def enqueueMessage(currentMessage):
        yield messageQueue.put(currentMessage)

    yield nc.subscribe("new_event", "", enqueueMessage)

@tornado.gen.coroutine
def main():
    tornado.ioloop.IOLoop.current().spawn_callback(consumer)
    yield producer()

if __name__ == '__main__':
    main()
    tornado.ioloop.IOLoop.current().start()

我的问题是：

1) 这是使用 Tornado 调用阻塞函数的正确方法吗？

2) 实施始终监听的消费者/生产者方案的最佳做法是什么？恐怕我的while True: 语句实际上阻塞了处理器……

3) 我如何检查队列以确保大量呼叫正在排队？我尝试过使用 Queue().qSize()，但它总是返回零，这让我想知道入队是否正确完成。

【问题讨论】：

注意 GIL 对 python 多线程的限制。如果可能的话，我会使用 PorcessPoolExceutor，例如 stackoverflow.com/questions/33553940/…
这是一个有趣的方法...我不知道 ProcessPoolExecutor 有它自己的队列。 ThreadPoolExecutor 也是如此吗？我也有兴趣最多运行一名工作人员，因为必须按顺序处理事件......在这种情况下，GIL 限制也会影响我吗？

标签： python multithreading tornado

【解决方案1】：

一般规则（感谢 NYKevin）是：

用于 CPU 和 GPU 绑定计算的多处理。
用于非阻塞 I/O 的事件驱动的东西（在可能的情况下应该优先于阻塞 I/O，因为它可以更有效地扩展）。
阻塞 I/O 的线程（您也可以使用多处理，但每个进程的开销可能不值得）。

ThreadPoolExecutor 用于 IO，ProcessPoolExecutor 用于 CPU。两者都有内部队列，都最多扩展到指定max_workers。更多关于concurrent executors in docs的信息。

所以答案是：

重新实现池是开销。线程或进程取决于您打算做什么。
while True 不会阻塞，例如一些产生了异步调用（甚至yield gen.sleep(0.01)），它将控制权交还给ioloop
qsize() 是打电话的权利，但由于我没有运行/调试这个，我会采取不同的方法（现有池），所以很难在这里找到问题。

【讨论】：