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. Выполнить действие A n раз на основе значения, определенного в request.body ( одновременно, если возможно)

Раньше для выполнения действия A n раз я использовал threading.ThreadPoolExecutor, но я не уверен, как мне правильно это обрабатывать с веб-сервером. работает параллельно.

Моя идея была примерно такой:

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)

Теперь этот фрагмент кода, конечно, блокируется, и я надеялся, что веб-сервер продолжит работать параллельно, в то время как я запускаю while-l oop.

Возможна ли такая установка? Может быть, мне нужно использовать другие модули? Любая помощь с благодарностью!

Answer 1

Работает не так, как ожидалось, потому что time.sleep является функцией блокировки.

Вместо использования списка и некоторого времени l oop и режима ожидания для проверки наличия новых элементов в списке, используйте queues.Queue от Tornado, который позволит вам асинхронно проверять наличие новых элементов. ,

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)