从asyncio中的两个协同程序中选择第一个结果

Question

题

使用Python的asyncio模块,如何从多个协同程序中选择第一个结果？

例

我可能想在队列上等待时实现超时:

result = yield from select(asyncio.sleep(1),
                           queue.get())

类似的行动

这与Goselect或Clojurecore.async.alt!类似.它就像是反过来asyncio.gather(聚集就像all,选择就像any.)

Answer 1

简单的解决方案,使用asyncio.wait及其FIRST_COMPLETED参数:

import asyncio

async def something_to_wait():
    await asyncio.sleep(1)
    return "something_to_wait"

async def something_else_to_wait():
    await asyncio.sleep(2)
    return "something_else_to_wait"


async def wait_first():
    done, pending = await asyncio.wait(
        [something_to_wait(), something_else_to_wait()],
        return_when=asyncio.FIRST_COMPLETED)
    print("done", done)
    print("pending", pending)

asyncio.get_event_loop().run_until_complete(wait_first())

得到:

done {<Task finished coro=<something_to_wait() done, defined at stack.py:3> result='something_to_wait'>}
pending {<Task pending coro=<something_else_to_wait() running at stack.py:8> wait_for=<Future pending cb=[Task._wakeup()]>>}
Task was destroyed but it is pending!
task: <Task pending coro=<something_else_to_wait() running at stack.py:8> wait_for=<Future pending cb=[Task._wakeup()]>>

Answer 2

在想超时应用到任务的情况下，有一个标准库函数正是这样做的：asyncio.wait_for()。你的例子可以这样写：

try:
  result = await asyncio.wait_for(queue.get(), timeout=1)
except asyncio.TimeoutError:
  # This block will execute if queue.get() takes more than 1s.
  result = ...

但这仅适用于超时的特定情况。这里的其他两个答案概括为任意一组任务，但这些答案都没有显示如何清理未首先完成的任务。这就是导致输出中出现“任务已销毁但正在挂起”消息的原因。在实践中，您应该对那些挂起的任务做一些事情。根据您的示例，我假设您不关心其他任务的结果。下面是一个wait_first()函数示例，该函数返回第一个已完成任务的值并取消其余任务。

import asyncio, random

async def foo(x):
    r = random.random()
    print('foo({:d}) sleeping for {:0.3f}'.format(x, r))
    await asyncio.sleep(r)
    print('foo({:d}) done'.format(x))
    return x

async def wait_first(*futures):
    ''' Return the result of the first future to finish. Cancel the remaining
    futures. '''
    done, pending = await asyncio.wait(futures,
        return_when=asyncio.FIRST_COMPLETED)
    gather = asyncio.gather(*pending)
    gather.cancel()
    try:
        await gather
    except asyncio.CancelledError:
        pass
    return done.pop().result()

async def main():
    result = await wait_first(foo(1), foo(2))
    print('the result is {}'.format(result))

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

运行这个例子：

# export PYTHONASYNCIODEBUG=1
# python3 test.py
foo(1) sleeping for 0.381
foo(2) sleeping for 0.279
foo(2) done
the result is 2
# python3 test.py
foo(1) sleeping for 0.048
foo(2) sleeping for 0.515
foo(1) done
the result is 1
# python3 test.py
foo(1) sleeping for 0.396
foo(2) sleeping for 0.188
foo(2) done
the result is 2

没有关于挂起任务的错误消息，因为每个挂起任务都已正确清理。

在实践中，你可能想要wait_first()返回未来，而不是未来的结果，否则试图弄清楚哪个未来完成了会很混乱。但在这里的例子中，我返回了未来的结果，因为它看起来更干净一些。

Answer 3

您可以同时使用实现这个asyncio.wait和asyncio.as_completed:

import asyncio

@asyncio.coroutine
def ok():
    yield from asyncio.sleep(1)
    return 5

@asyncio.coroutine
def select1(*futures, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    return (yield from next(asyncio.as_completed(futures)))

@asyncio.coroutine
def select2(*futures, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    done, running = yield from asyncio.wait(futures,
                                            return_when=asyncio.FIRST_COMPLETED)
    result = done.pop()
    return result.result()

@asyncio.coroutine
def example():
    queue = asyncio.Queue()
    result = yield from select1(ok(), queue.get())
    print('got {}'.format(result))
    result = yield from select2(queue.get(), ok())
    print('got {}'.format(result))

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(example())

输出:

got 5
got 5
Task was destroyed but it is pending!
task: <Task pending coro=<get() done, defined at /usr/lib/python3.4/asyncio/queues.py:170> wait_for=<Future pending cb=[Task._wakeup()]> cb=[as_completed.<locals>._on_completion() at /usr/lib/python3.4/asyncio/tasks.py:463]>
Task was destroyed but it is pending!
task: <Task pending coro=<get() done, defined at /usr/lib/python3.4/asyncio/queues.py:170> wait_for=<Future pending cb=[Task._wakeup()]>>

两个实现都返回第一个完成后产生的值Future,但您可以轻松调整它以返回Future自身.请注意,因为Future传递给每个select实现的其他实现永远不会产生,所以当进程退出时会引发警告.