Tornado asynchronous job in tornado gen coroutine

Question

Я написал приложение, которое выполняет задание из очереди и выполняет его асинхронно.

def job(self): 
    print 'In job' 
    time.sleep(0.01) 

@gen.coroutine 
def start_jobs(self): 
    jobs = filter(lambda x: x['status'] == 0, self.queue) 
    for job in jobs: 
     yield self.job() 
    print 'exit from start job' 

Но этот код не работает.

Выход:

В работе

В работе

В работе и т.д.

Как это сделать правильно?

Как мне заставить работать с фьючерсами, и есть ли более простой способ сделать это с помощью Tornado?

Answer 1

Никогда не звоните time.sleep в торнадо! Вместо этого используйте yield gen.sleep.

Установка Торо с pip install toro и использовать JoinableQueue:

import random 
from tornado import ioloop, gen 
import toro 


class C(object): 
    def __init__(self): 
     self.queue = toro.JoinableQueue() 

    @gen.coroutine 
    def start_jobs(self): 
     while True: 
      job_id = yield self.queue.get() 
      self.job(job_id) 

    @gen.coroutine 
    def job(self, job_id): 
     print 'job_id', job_id 
     yield gen.sleep(random.random()) 
     print 'job_id', job_id, 'done' 
     self.queue.task_done() 


c = C() 
for i in range(5): 
    c.queue.put_nowait(i) 

c.start_jobs() 

io_loop = ioloop.IOLoop.instance() 

# block until all tasks are done 
c.queue.join().add_done_callback(lambda future: io_loop.stop()) 
io_loop.start() 

Начиная с Tornado 4.2, Торо является частью Торнадо, так что вы можете просто сделать queue = tornado.queues.Queue() вместо того, чтобы использовать в Toro JoinableQueue:

http://tornado.readthedocs.org/en/latest/releases/v4.2.0.html#new-modules-tornado-locks-and-tornado-queues