Адаптация celery.task.http.URL для торнадо

Сельдерей включает в себя модуль, который может создавать асинхронные HTTP-запросы с использованием amqp или какого-либо другого ферментера. Я использую tornado-celery производителя для асинхронной публикации сообщений. Как я понял, tornado-celery использует pika для этого. Вопрос в том, как адаптировать celery.task.http.URL для торнадо (сделать его неблокирующим). В основном есть два места, которые нужно уточнить:

  • HttpDispatch.make_request() должны быть реализованы с использованием tornado async http client;

  • URL.get_async(**kw) или URL.post_async(**kw) должен быть переопределен соответствующим неблокирующим кодом с использованием API торнадо. Например:

    class NonBlockingURL(celery.task.http.URL):

    @gen.coroutine
    def post_async(self, **kwargs):
        async_res = yield gen.Task(self.dispatcher.delay, 
                                   str(self), 'POST', **kwargs)
        raise gen.Return(async_res)

Но я не мог понять, как это сделать правильно и лаконично. Как сделать его полностью неблокирующим как асинхронным? Кстати, я использую бэкэнд amqp.

Пожалуйста, предоставьте мне хороший ориентир или даже лучше, пример.

Ответы

Ответ 1

Фактически, вам нужно решить, используете ли вы метод async для Tornado или используете очередь, подобную элементу. Нет смысла использовать оба варианта, потому что очередь быстро отвечает за состояние очереди, поэтому нет смысла торнадо делать что-то еще, ожидая ответа очереди. Чтобы решить между двумя решениями, я бы сказал:

Сельдерей: более модульный, простой для распространения на разных ядрах или разных машинах, задача может быть использована другими людьми, кроме торнадо, вам необходимо установить и продолжить работу softare (amqp, celler workers...)

Async in Tornado: более монолитный, одна программа делает все, более короткий код, одну программу для запуска

Чтобы использовать асинхронный метод Tornado, обратитесь к документации. Вот короткое решение, использующее сельдерей и торнадо:

task.py

 from celery import Celery,current_task
 import time
 celery=Celery('tasks',backend='amqp',result_backend='amqp')

 @celery.task
 def MyTask(url,resid):
     for i in range(10):
         time.sleep(1)
         current_task.update_state(state='running',meta={'i': i})
     return 'done'

server.py

 import tasks
 from tornado import RequestHandler,....
 from tornado.web import Application
 dictasks={}
 class runtask(RequestHandler):
     def post(self):
         i=len(dictasks)
         dictasks[i]=task.MyTask.delay()
             self.write(i)

 class chktask(RequestHandler):
     def get(self,i):
         i=int(i)
         if dictasks[i].ready():
             self.write(dictasks[i].result)
             del dictasks[i]
         else:
             self.write(dictasks[i].state + ' i:' + dictasks[i].info.get('i',-1))


 Application = Application([
     (r"/runtask", runtask}),
     (r"/chktask/([0-9]+)", chktask),

 etc.