Rozważmy następujący przykład z dokumentacji:

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number+1):
        print("Task %s: Compute factorial(%s)..." % (name, i))
        await asyncio.sleep(1)
        f *= i
    print("Task %s: factorial(%s) = %s" % (name, number, f))

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
    factorial("A", 2),
    factorial("B", 3),
    factorial("C", 4),
))
loop.close()

Funkcja zbierania jest prezentowana jako taka w module:

asyncio.gather(*coros_or_futures, loop=None, return_exceptions=False)

Działa wszystko w porządku, ale dla mojego prawdziwego problemu życia muszę przejść w funkcji zbierania, a nie wielość funkcji z hardcodowanymi argumentami, ale raczej rozumieniem tuple niektórych formularzy tworzących wiele funkcji.

Próbowałem:

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
    [factorial(str(g),g) for g in range(3)]
))
loop.close()

Ale to nie działa. Każdy wie, jak zgromadzić funkcję, aby pracować z programowo utworzoną listą funkcji?

Być jasnym: funkcja gather nie jest przeze mnie definiowana, więc nie mogę usunąć * z jego definicji i po prostu przekazać listę argumentów takich. Musiałbym "rozpakować" listę, ale nie wiem, jak.

5
jim basquiat 25 czerwiec 2017, 13:39

3 odpowiedzi

Najlepsza odpowiedź

Znalazłem rozwiązanie:

loop.run_until_complete(asyncio.gather(*(factorial(str(g),g) for g in range(3)))
1
jim basquiat 25 czerwiec 2017, 10:57

Lista coroutines może być generowana dynamicznie i przekazywana w następujący sposób:

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({i})...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")

async def main():
    a = factorial("A", 2)
    b = factorial("B", 3)
    c = factorial("C", 4)
    #
    # example passing in list of coroutines
    #
    await asyncio.gather(*[a,b,c])

asyncio.run(main())
1
Mark Andersen 26 luty 2019, 20:54

Możesz także użyć itertools.starmap dla tego zadania:

import asyncio
import itertools

async def factorial(name, number):
    f = 1
    for i in range(2, number+1):
        print("Task %s: Compute factorial(%s)..." % (name, i))
        await asyncio.sleep(1)
        f *= i
    print("Task %s: factorial(%s) = %s" % (name, number, f))

loop = asyncio.get_event_loop()

args = [(str(g), g) for g in range(3)]
tasks = itertools.starmap(factorial, args)
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()

Zrób iterator, który oblicza funkcję przy użyciu argumentów uzyskanych z iterable. Używany zamiast mapy (), gdy parametry argumentów są już pogrupowane w Ktoczniki z jednego języka (dane (dane są "zapinane na zapięcie").

Aby uzyskać więcej czytania: Oto

4
Yuval Pruss 26 czerwiec 2017, 06:29