永远不会引发 asyncio.TimeoutError

Question

我想测试asyncio 如何处理阻塞进程。

我的代码一定有问题，因为 asyncio.TimeoutError 从未被提出：

import asyncio, random, time

q = asyncio.Queue()

MAX_WAIT = 5

@asyncio.coroutine
def blocking_task(sec):
    print('This task will sleep {} sec.'.format(sec))
    time.sleep(sec)

@asyncio.coroutine
def produce():
    while True:
        q.put_nowait(random.randint(1,10))
        yield from asyncio.sleep(0.5 + random.random())

@asyncio.coroutine
def consume():
    while True:
        value = yield from q.get()
        try:
            yield from asyncio.wait_for(blocking_task(value), MAX_WAIT)
        except asyncio.TimeoutError:
            print('~/~ Job has been canceled !!')
        else:
            print('=/= Job has been done :]')


loop = asyncio.get_event_loop()
asyncio.ensure_future(produce())
asyncio.ensure_future(consume())
loop.run_forever()

此代码产生以下输出：

$ ./tst3.py 
This task will sleep 2 sec.
=/= Job has been done :]
This task will sleep 1 sec.
=/= Job has been done :]
This task will sleep 7 sec.
=/= Job has been done :]

Answer 1

使用asyncio.sleep 而不是sleep

TimeoutError of asyncio 与 buildin TimeoutError 不同。这就是为什么你不能使用 time.sleep 来触发这个错误。在 asyncio.coroutine 中触发 TimeoutError，只能使用 asyncio 模块实现的 timer。

@asyncio.coroutine
def blocking_task(sec):
    print('This task will sleep {} sec.'.format(sec))
    yield from asyncio.sleep(sec)

结果

This task will sleep 10 sec.
~/~ Job has been canceled !!
This task will sleep 3 sec.
=/= Job has been done :]
This task will sleep 4 sec.
=/= Job has been done :]
This task will sleep 2 sec.
=/= Job has been done :]
This task will sleep 7 sec.
~/~ Job has been canceled !!
This task will sleep 2 sec.
=/= Job has been done :]