python - 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

Use asyncio.sleep instead of sleep

TimeoutError of asyncio与 buildin TimeoutError 不同。这就是为什么您不能使用 time.sleep 来触发此错误。要在asyncio.coroutine中触发TimeoutError，只能使用asyncio模块实现的定时器。

@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 :]