从异步中的两个协同项中选择第一个结果

Question

问题

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

示例

我可能希望在等待队列时实现超时：

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

坐骨神经手术

这将类似于select或core.async.alt!。它类似于asyncio.gather的逆流( the类似于all，select类似于any)。

Answer 1

您可以使用asyncio.wait和asyncio.as_completed实现这一点。

import asyncio


async def ok():
    await asyncio.sleep(1)
    return 5

async def select1(*futures, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    return (await next(asyncio.as_completed(futures)))

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

async def example():
    queue = asyncio.Queue()
    result = await select1(ok(), queue.get())
    print('got {}'.format(result))
    result = await 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本身。请注意，由于传递给每个select实现的另一个select从未产生过，因此在进程退出时会引发警告。

Answer 2

简单的解决方案，通过使用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 3

在想要对任务应用超时的情况下，有一个标准的库函数，它就是这样做的：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()返回未来，而不是未来的结果，否则，试图找出哪个未来完成将非常令人困惑。但是在这里的例子中，我返回了未来的结果，因为它看起来有点干净。