python asyncio add_done_callback with async def

Question

我有两个函数：第一个，def_a，是一个异步函数，第二个是def_b，它是一个常规函数，调用的结果是def_a作为add_done_callback函数的回调。

我的代码如下所示：

import asyncio
def def_b(result):
next_number = result.result()
# some work on the next_number
print(next_number + 1)
async def def_a(number):
await some_async_work(number)
return number + 1
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(def_a(1))
task.add_done_callback(def_b)
response = loop.run_until_complete(task)
loop.close()

而且它的工作非常完美。

当第二个函数def_b变得异步时，问题就开始了。现在它看起来像这样：

async def def_b(result):
next_number = result.result()
# some asynchronous work on the next_number
print(next_number + 1)

但是现在我无法将其提供给add_done_callback函数，因为它不是常规函数。

我的问题是 - 如果def_b异步，是否有可能以及如何为add_done_callback函数提供def_b？

Answer 1

add_done_callback

被认为是一个"低级"接口。 使用协程时，可以通过多种方式链接它们，例如：

import asyncio

async def my_callback(result):
print("my_callback got:", result)
return "My return value is ignored"

async def coro(number):
await asyncio.sleep(number)
return number + 1

async def add_success_callback(fut, callback):
result = await fut
await callback(result)
return result

loop = asyncio.get_event_loop()
task = asyncio.ensure_future(coro(1))
task = add_success_callback(task, my_callback)
response = loop.run_until_complete(task)
print("response:", response)
loop.close()

请记住add_done_callback如果你的未来引发异常，它仍然会调用回调(但调用result.result()会引发它)。

Answer 2

这仅适用于一个未来的作业，如果你有多个异步作业，它们会互相阻塞，更好的方法是使用 asyncio.as_completed() 迭代未来的列表：

import asyncio
async def __after_done_callback(future_result):
# await for something...
pass
async def __future_job(number):
await some_async_work(number)
return number + 1
loop = asyncio.get_event_loop()
tasks = [asyncio.ensure_future(__future_job(x)) for x in range(100)]  # create 100 future jobs
for f in asyncio.as_completed(tasks):
result = await f
await __after_done_callback(result)
loop.close()

Answer 3

你可以试试aiodag库。这是一个非常轻量级的asyncio包装器，它抽象掉了一些你通常必须考虑的异步管道。从此示例中，您将无法判断事物是否异步运行，因为它只是依赖于另一个任务的 1 个任务，但它都是异步运行的。

import asyncio
from aiodag import task
@task
async def def_b(result):
# some asynchronous work on the next_number
print(result + 1)
@task
async def def_a(number):
await asyncio.sleep(number)
return number + 1
async def main():
a = def_a(1)
b = def_b(a)  # this makes task b depend on task a
return await b
loop = asyncio.get_event_loop()
asyncio.set_event_loop(loop)
response = loop.run_until_complete(main())