python-3.x - Python3如何asyncio.gather()部分函数列表

Question

我正在尝试创建可以传递给asyncio.gather()的协程列表

但是，我想在将这些协程添加到列表时将参数附加到这些协程。

我下面显示的当前方法使用functools.partial。不幸的是asyncio.gather不接受部分函数，​​这很有意义。

对我来说没有意义的是如何找到解决方案。

示例代码:

async def test(arg1):
    print(arg1)

statements = []

function = functools.partial(test, "hello world")
statements.append(function)

results = await asyncio.gather(*statements)

那么，如何将参数附加到函数，以便仍可以将其传递给asyncio.gather？

*编辑

看来我很傻。

我的解决方案非常简单，不要使用functools.partial，而只是将协程直接添加到列表中。

代码:

async def test(arg1):
    print(arg1)

async def main():
    statements = []
    statements.append(test("hello_world"))
    results = await asyncio.gather(*statements)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Answer 1

仅一次收集多个异步函数调用的结果
如果只需要运行一次async函数调用序列，则可以简单地将其立即返回的协程对象存储在列表中，将解压后的列表传递给asyncio.gather，然后传递await以从协程中收集最终返回值。

async def upper_cased(value: str) -> str:
    await asyncio.sleep(1)
    return value.upper()

coroutines = [
    upper_cased("h"),
    upper_cased("e"),
    upper_cased("l"),
    upper_cased("l"),
    upper_cased("o"),
    upper_cased(" "),
    upper_cased("w"),
    upper_cased("o"),
    upper_cased("r"),
    upper_cased("l"),
    upper_cased("d"),
]
print("".join(await asyncio.gather(*coroutines)))
# prints HELLO WORLD

之所以可以这样做是因为与普通函数不同，调用 async函数(不等待)不会开始执行它，它只是返回一个对象，该对象包含函数调用的所有必要细节，包括参数值。以后可以使用此对象以不同的方式完成其执行，其中之一是使用 await。
多次收集多个异步函数调用的结果
如果您需要多次运行相同的 async函数序列(带有预定参数) 多次，则仅将它们的返回值存储在列表中，然后将解压缩后的列表传递给 asyncio.gather 将不起作用。
这是因为协程对象只能等待一次。当它们已经返回值时再次等待它们将引发 RuntimeError。

async def random_cased(value: str) -> str:
    await asyncio.sleep(1)
    if random.randint(0, 1) == 1:
        return value.upper()
    return value.lower()

coroutines = [
    random_cased("h"),
    random_cased("e"),
    random_cased("l"),
    random_cased("l"),
    random_cased("o"),
    random_cased(" "),
    random_cased("w"),
    random_cased("o"),
    random_cased("r"),
    random_cased("l"),
    random_cased("d"),
]
print("".join(await asyncio.gather(*coroutines)))
# prints HeLlO wOrLd (or any other random cased variation)

print("".join(await asyncio.gather(*coroutines)))
# raises RuntimeError (cannot reuse already awaited coroutine)

对于这种情况，可以使用 functools.partial对象存储 async函数及其参数值。

partial_funcs = [
    functools.partial(random_cased, "h"),
    functools.partial(random_cased, "e"),
    functools.partial(random_cased, "l"),
    functools.partial(random_cased, "l"),
    functools.partial(random_cased, "o"),
    functools.partial(random_cased, " "),
    functools.partial(random_cased, "w"),
    functools.partial(random_cased, "o"),
    functools.partial(random_cased, "r"),
    functools.partial(random_cased, "l"),
    functools.partial(random_cased, "d"),
]
print("".join(await asyncio.gather(*[func() for func in partial_funcs])))
# prints HeLlO wOrLd (or any other random cased variation)

print("".join(await asyncio.gather(*[func() for func in partial_funcs])))
# prints HeLlO wOrLd (or any other random cased variation)

print("".join(await asyncio.gather(*[func() for func in partial_funcs])))
# prints HeLlO wOrLd (or any other random cased variation)

如果需要，还可以使用其他容器代替 functools.partial。例如，一个用于存储函数的列表和另一个用于存储其参数的列表。