Запуск фоновой задачи async в Tornado

Читая документацию Tornado, очень ясно, как вызвать функцию async для ответа:

class GenAsyncHandler(RequestHandler):
    @gen.coroutine
    def get(self):
        http_client = AsyncHTTPClient()
        response = yield http_client.fetch("http://example.com")
        do_something_with_response(response)
        self.render("template.html")

Не хватает того, как вызов должен выполняться асинхронно для фоновой задачи, которая не имеет отношения к текущему запросу:

class GenAsyncHandler(RequestHandler):
    @gen.coroutine
    def _background_task():
        pass  # do lots of background stuff

    @gen.coroutine
    def get(self):
        _dont_care = yield self._background_task()
        self.render("template.html")

Этот код должен работать, за исключением того, что он выполняется синхронно, и запрос ожидает его до его завершения.

Каков правильный способ асинхронного вызова этой задачи, сразу же возвращая текущий запрос?

Ответ 1

Обновление: Начиная с Tornado 4.0 (июль 2014 г.), нижеприведенная функциональность доступна в методе IOLoop.spawn_callback.

К сожалению, это довольно сложно. Вам нужно как отсоединить фоновую задачу от текущего запроса (чтобы сбой в фоновой задаче не привел к случайному исключению, выданному в запросе), так и убедиться, что что-то прослушивает результат фоновой задачи (для регистрации ошибок). если ничего другого). Это означает что-то вроде этого:

from tornado.ioloop import IOLoop
from tornado.stack_context import run_in_stack_context, NullContext
IOLoop.current().add_future(run_in_stack_context(NullContext(), self._background_task),
                            lambda f: f.result())

Нечто подобное может быть добавлено к самому торнадо в будущем.

Ответ 2

Я рекомендую использовать toro. Он обеспечивает относительно простой механизм настройки фоновой очереди задач.

Следующий код (например, в queue.py) запускает простой "worker()", который просто ждет, пока в его очереди не будет чего-то. Если вы вызываете queue.add(function,async,*args,**kwargs), это добавляет элемент в очередь, который будет пробуждать рабочий(), который затем запускает задачу.

Я добавил параметр async, чтобы он мог поддерживать фоновые задачи, завернутые в @gen.coroutine, и те, у кого нет.

import toro,tornado.gen
queue = toro.Queue()
@tornado.gen.coroutine
def add(function,async,*args,**kwargs):
   item = dict(function=function,async=async,args=args,kwargs=kwargs)
   yield queue.put(item)

@tornado.gen.coroutine
def worker():
   while True:
      print("worker() sleeping until I get next item")
      item = yield queue.get()
      print("worker() waking up to process: %s" % item)
      try:
         if item['async']:
            yield item['function'](*item['args'],**item['kwargs'])
         else:
            item['function'](*item['args'],**item['kwargs'])
      except Exception as e:
         print("worker() failed to run item: %s, received exception:\n%s" % (item,e))

@tornado.gen.coroutine
def start():
   yield worker()

В главном приложении торнадо:

import queue
queue.start()

И теперь вы можете просто запланировать заднюю задачу:

def my_func(arg1,somekwarg=None):
   print("in my_func() with %s %s" % (arg1,somekwarg))

queue.add(my_func,False,somearg,somekwarg=someval)

Ответ 3

У меня есть трудоемкая задача в почтовом запросе, возможно, больше 30 минут, но клиент должен немедленно вернуть результат.

Во-первых, я использовал IOLoop.current(). spawn_callback. Оно работает! но! Если первая задача запроса запущена, вторая задача запроса заблокирована! Поскольку все задачи находятся в главном цикле событий при использовании spawn_callback, поэтому одна задача является синхронным исполнением, другие задачи блокированы.

Наконец, я использую tornado.concurrent. Пример:

import datetime
import time

from tornado.ioloop import IOLoop
import tornado.web
from tornado import concurrent

executor = concurrent.futures.ThreadPoolExecutor(8)


class Handler(tornado.web.RequestHandler):

    def get(self):
        def task(arg):
            for i in range(10):
                time.sleep(1)
                print(arg, i)

        executor.submit(task, datetime.datetime.now())
        self.write('request accepted')


def make_app():
    return tornado.web.Application([
        (r"/", Handler),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8000, '0.0.0.0')
    IOLoop.current().start()

и посетите http://127.0.0.1:8000, вы можете видеть, что он работает нормально:

2017-01-17 22:42:10.983632 0
2017-01-17 22:42:10.983632 1
2017-01-17 22:42:10.983632 2
2017-01-17 22:42:13.710145 0
2017-01-17 22:42:10.983632 3
2017-01-17 22:42:13.710145 1
2017-01-17 22:42:10.983632 4
2017-01-17 22:42:13.710145 2
2017-01-17 22:42:10.983632 5
2017-01-17 22:42:16.694966 0
2017-01-17 22:42:13.710145 3
2017-01-17 22:42:10.983632 6
2017-01-17 22:42:16.694966 1
2017-01-17 22:42:13.710145 4
2017-01-17 22:42:10.983632 7
2017-01-17 22:42:16.694966 2
2017-01-17 22:42:13.710145 5
2017-01-17 22:42:10.983632 8
2017-01-17 22:42:16.694966 3
2017-01-17 22:42:13.710145 6
2017-01-17 22:42:19.790646 0
2017-01-17 22:42:10.983632 9
2017-01-17 22:42:16.694966 4
2017-01-17 22:42:13.710145 7
2017-01-17 22:42:19.790646 1
2017-01-17 22:42:16.694966 5
2017-01-17 22:42:13.710145 8
2017-01-17 22:42:19.790646 2
2017-01-17 22:42:16.694966 6
2017-01-17 22:42:13.710145 9
2017-01-17 22:42:19.790646 3
2017-01-17 22:42:16.694966 7
2017-01-17 22:42:19.790646 4
2017-01-17 22:42:16.694966 8
2017-01-17 22:42:19.790646 5
2017-01-17 22:42:16.694966 9
2017-01-17 22:42:19.790646 6
2017-01-17 22:42:19.790646 7
2017-01-17 22:42:19.790646 8
2017-01-17 22:42:19.790646 9

Хотите помочь всем!

Ответ 4

Просто выполните:

self._background_task()

Строка _background_task возвращает a Future, которая не разрешена до завершения сопрограммы. Если вы не получаете Future и вместо этого просто выполняете следующую строку сразу, то get() не ждет завершения _background_task.

Интересная деталь заключается в том, что до завершения _background_task она поддерживает ссылку на self. (Не забудьте добавить self в качестве параметра, кстати.) Ваш RequestHandler не будет собираться мусором, пока не завершится _background_task.