Tornado HTTPServer 在收到 POST 请求时将对象添加到队列中

Question

我想创建一个网络服务器，在收到 POST 请求时自动处理“订单”。

到目前为止，我的代码如下所示：

from json import loads
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.web import Application, url, RequestHandler


order_list = list()


class MainHandler(RequestHandler):  

    def get(self):
        pass

    def post(self):
        if self.__authenticate_incoming_request():
            payload = loads(self.request.body)
            order_list.append(payload)
        else:
            pass

    def __authenticate_incoming_request(self) -> bool:
        # some request authentication code
        return True


def start_server():
    application = Application([
        url(r"/", MainHandler)
    ])

    server = HTTPServer(application)
    server.listen(8080)

    IOLoop.current().start()


if __name__ == '__main__':
    start_server()

这是我想要实现的目标：

1. 接收包含传入“订单”信息的 POST 请求
2. 根据request.body 中定义的值执行操作A n 次（如果可能，同时执行）

以前，为了执行操作A n 次，我使用了 threading.ThreadPoolExecutor，但我不确定如何正确处理并行运行的 Web 服务器。

我的想法是这样的：

start_server()

tpe = ThreadPoolExecutor(max_workers=10)
while True:
     if order_list:
         new_order = order_list.pop(0)
         tpe.submit(my_action, new_order)         # my_action is my desired function

     sleep(5)

现在这段代码当然是阻塞的，我希望 Web 服务器在我运行 while 循环的同时继续并行运行。

这样的设置可能吗？我可能需要使用其他模块吗？非常感谢任何帮助！

Answer 1

它没有按预期工作，因为time.sleep 是一个阻塞函数。

使用 Tornado 的 queues.Queue 可以异步检查新项目，而不是使用列表和 while 循环和休眠来检查列表中的新项目。

from tornado.queues import Queue

order_queue = Queue()
tpe = ThreadPoolExecutor(max_workers=10)

async def queue_consumer():
    # The old while-loop is now converted into a coroutine
    # and an async for loop is used instead
    async for new_order in order_queue:
        # run your function in threadpool
        IOLoop.current().run_in_executor(tpe, my_action, new_order)


def start_server():
   # ...
   # run queue_consumer function before starting the loop
   IOLoop.current().spawn_callback(queue_consumer)
   IOLoop.current().start()

像这样将项目放入队列中：

order_queue.put(payload)