python - Tornado 异步队列不等待

Question

我从这个 Tornado documentation 修改了带有生产者和消费者的示例队列，但似乎传递给 get() 的超时参数根本不起作用，因为消费者在抛出异常之前不会等待 10 秒。理想情况下，生产者和消费者将同时运行。还有，我不知道传入的timeout参数是秒还是毫秒:

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.queues import Queue

q = Queue()

@gen.coroutine
def consumer():
    try:
        while True:
            item = yield q.get(timeout=10000)
            try:
                print('Doing work on %s' % item)      
            finally:
                q.task_done()
    except gen.TimeoutError:
        print('timeout')
        return

@gen.coroutine
def producer():
    for item in range(5):
        yield q.put(item)
        print('Put %s' % item)
        yield gen.sleep(2)

@gen.coroutine
def main():
    # Start consumer without waiting (since it never finishes).
    IOLoop.current().spawn_callback(consumer)
    yield producer()     # Wait for producer to put all tasks.
    yield q.join()       # Wait for consumer to finish all tasks.
    print('Done')

IOLoop.current().run_sync(main)

这是它的执行:

Put 0
Doing work on 0
timeout
Put 1
Put 2
Put 3
Put 4

Answer 1

超时

正如您在 Tornado' Queue.get docs 中所读到的那样:

Returns a Future which resolves once an item is available or raises tornado.gen.TimeoutError after a timeout.

但这可能会产生误导，因为 timeout实际上是一个 deadline .所以它必须用 datetime.timedelta object 指定:

import datetime
yield q.get(timeout=datetime.timedelta(seconds=1))

或绝对时间:

timeout = 1.5  # in seconds, floats acceptable
deadline = IOLoop.current().time() + timeout
# in most cases IOLoop time is just time.time()
# I've used separate variables only for the notion

yield q.get(timeout=deadline)

在Toro ，已合并到 Tornado 中，此参数称为 deadline .

在您的代码中，您指定了超时 10000 ，这意味着到 Thu, 01 Jan 1970 02:46:40 GMT 的截止日期.

消费者循环

既然你有try/except阻塞整个函数，包括 while循环，当TimeoutError发生您的消费者停止工作。将异常处理移至 while循环。

工作示例:

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.queues import Queue

q = Queue()

@gen.coroutine
def consumer():
    i = 0
    while True:
        i += 1
        print('get cycle %s' % i)
        try:
            item = yield q.get(IOLoop.instance().time() + 3)
            try:
                print('Doing work on %s' % item)
            finally:
                q.task_done()
        except gen.TimeoutError:
            print('timeout')

@gen.coroutine
def producer():
    for item in range(5):
        yield q.put(item)
        print('Put %s' % item)
        yield gen.sleep(2)

@gen.coroutine
def main():
    # Start consumer without waiting (since it never finishes).
    IOLoop.current().spawn_callback(consumer)
    yield producer()     # Wait for producer to put all tasks.
    yield q.join()       # Wait for consumer to finish all tasks.
    print('Done')