遍历 asyncio.coroutine

Question

最近我在玩 asyncio，虽然我开始对它的工作原理有了一些直觉，但有些事情我还是搞不明白。我不确定是我构造的方式不对，还是我想做的事情本身就不合理。

简单来说，我想要能够遍历一个返回值的 asyncio.coroutine。比如，我想做类似这样的事情：

@asyncio.coroutine
def countdown(n):
    while n > 0:
        yield from asyncio.sleep(1)
        n = n - 1
        yield n

@asyncio.coroutine
def do_work():
    for n in countdown(5):
        print(n)

loop.run_until_complete(do_work())

但是，这样做会从 asyncio 的深处抛出一个异常。我试过其他方法，比如 for n in (yield from countdown(5)): ...，但这也会出现类似的难以理解的运行时异常。

我一时看不出为什么不应该这样做，但我感觉自己快到理解的极限了。

所以：

• 如果可以这样做，我该怎么做？
• 如果不可以，为什么不可以？

如果这个问题不清楚，请告诉我！

Answer 1

更新：看起来 Python 3.5 对这个功能的支持更好：

我也遇到了同样的问题，并且受到 aio-s3 代码的启发，觉得应该有更优雅的解决方案。

import asyncio

def countdown(number):
    @asyncio.coroutine
    def sleep(returnvalue):
        yield from asyncio.sleep(1)
        return returnvalue
    for n in range(number, 0, -1):
        yield sleep(n)

@asyncio.coroutine
def print_countdown():
    for future in countdown(5):
        n = yield from future
        print ("Counting down: %d" % n)

asyncio.get_event_loop().run_until_complete(print_countdown())

解释：countdown 方法会生成一些未来的结果，每个结果在休眠1秒后会变成你提供的数字。

print_countdown 函数会取第一个未来的结果，使用 yield from 来等待它完成（这会暂停，直到结果出来），然后得到想要的结果：n。

Answer 2

在Python 3.5中，新增了async for这种写法。不过，异步迭代器的函数写法还没有出现（也就是说，在async函数里不能用yield）。这里有个解决办法：

import asyncio
import inspect

class escape(object):
    def __init__(self, value):
        self.value = value

class _asynciter(object):
    def __init__(self, iterator):
        self.itr = iterator
    async def __aiter__(self):
        return self
    async def __anext__(self):
        try:
            yielded = next(self.itr)
            while inspect.isawaitable(yielded):
                try:
                    result = await yielded
                except Exception as e:
                    yielded = self.itr.throw(e)
                else:
                    yielded = self.itr.send(result)
            else:
                if isinstance(yielded, escape):
                    return yielded.value
                else:
                    return yielded
        except StopIteration:
            raise StopAsyncIteration

def asynciter(f):
    return lambda *arg, **kwarg: _asynciter(f(*arg, **kwarg))

然后你的代码可以写成：

@asynciter
def countdown(n):
    while n > 0:
        yield from asyncio.sleep(1)
        #or:
        #yield asyncio.sleep(1)
        n = n - 1
        yield n

async def do_work():
    async for n in countdown(5):
        print(n)

asyncio.get_event_loop().run_until_complete(do_work())

想了解这种新写法，以及这段代码是怎么工作的，可以查看PEP 492

Answer 3

在使用 asyncio 的协程时，你应该使用 yield from，而不是 yield。这是设计上的要求。yield from 后面应该接另一个协程或者 asyncio.Future 实例。

调用协程本身时，也应该使用 yield from，比如 yield from countdown(5)。

对于你的情况，我建议使用队列：

import asyncio

@asyncio.coroutine
def countdown(n, queue):
    while n > 0:
        yield from asyncio.sleep(1)
        n = n - 1
        yield from queue.put(n)
    yield from queue.put(None)

@asyncio.coroutine
def do_work():
    queue = asyncio.Queue()
    asyncio.async(countdown(5, queue))
    while True:
        v = yield from queue.get()
        if v:
            print(v)
        else:
            break

asyncio.get_event_loop().run_until_complete(do_work())

你可以检查 countdown 返回的值，下面的例子是可以工作的。但我觉得这样做不是个好习惯：

1. 很容易搞乱

2. 你还是不能把 countdown 的调用和其他函数，比如 itertools 的函数组合在一起。我是说像 sum(countdown(5)) 或者 itertools.accumulate(countdown(5)) 这样的。

总之，这里有一个混合使用 yield 和 yield from 的协程例子：

import asyncio

@asyncio.coroutine
def countdown(n):
    while n > 0:
        yield from asyncio.sleep(1)
        n = n - 1
        yield n

@asyncio.coroutine
def do_work():
    for n in countdown(5):
        if isinstance(n, asyncio.Future):
            yield from n
        else:
            print(n)

asyncio.get_event_loop().run_until_complete(do_work())