Generator erstellen, der Coroutine-Ergebnisse liefert, wenn die Koroutinen beendet werden

Question

Derzeit habe ich einen ineffizienten synchronen Generator, der viele HTTP-Anfragen nacheinander absetzt und die Ergebnisse liefert. Ich möchte asyncio und aiohttp verwenden, um die Anfragen parallel zu machen und dadurch diesen Generator zu beschleunigen, aber ich möchte es als ein gewöhnlicher Generator behalten (nicht PEP 525 async generator), damit der Nicht-Async-Code, der es aufruft, nicht muss modifiziert sein. Wie kann ich einen solchen Generator erstellen?

Answer 1

asyncio.as_completed(), derzeit kaum dokumentiert, nimmt eine iterable von Coroutines oder Futures und gibt eine iterable von Futures in der Reihenfolge, dass die Input-Futures abgeschlossen. Normalerweise, würden Sie eine Schleife über das Ergebnis und await die Mitglieder aus dem Innern einer async Funktion ...

import asyncio 

async def first(): 
    await asyncio.sleep(5) 
    return 'first' 

async def second(): 
    await asyncio.sleep(1) 
    return 'second' 

async def third(): 
    await asyncio.sleep(3) 
    return 'third' 

async def main(): 
    for future in asyncio.as_completed([first(), second(), third()]): 
     print(await future) 

loop = asyncio.get_event_loop() 

# Prints 'second', then 'third', then 'first' 
loop.run_until_complete(main()) 

... aber für die Zwecke dieser Frage, wollen, was wir liefern zu können, ist zu diese Ergebnisse stammen von einem gewöhnlichen Generator, so dass normaler synchroner Code sie verbrauchen kann, ohne jemals zu wissen, dass async Funktionen unter der Haube verwendet werden. Wir können das tun, indem loop.run_until_complete() an den Termin von unseren as_completed Anruf ergab Aufruf ...

import asyncio 

async def first(): 
    await asyncio.sleep(5) 
    return 'first' 

async def second(): 
    await asyncio.sleep(1) 
    return 'second' 

async def third(): 
    await asyncio.sleep(3) 
    return 'third' 

def ordinary_generator(): 
    loop = asyncio.get_event_loop() 
    for future in asyncio.as_completed([first(), second(), third()]): 
     yield loop.run_until_complete(future) 

# Prints 'second', then 'third', then 'first' 
for element in ordinary_generator(): 
    print(element) 

Auf diese Weise können wir unsere Asynchron-Code nicht-Asynchron-Land in einer Weise ausgesetzt haben, die nicht benötigt Anrufer, um irgendwelche Funktionen wie async zu definieren, oder sogar zu wissen, dass ordinary_generatorasyncio unter der Haube verwendet.

Als eine alternative Implementierung von ordinary_generator(), die mehr Flexibilität in einigen Fällen bietet, können wir immer wieder asyncio.wait() mit der FIRST_COMPLETED Flagge rufen statt Schleifen über as_completed():

import concurrent.futures 

def ordinary_generator(): 
    loop = asyncio.get_event_loop() 
    pending = [first(), second(), third()] 
    while pending: 
     done, pending = loop.run_until_complete(
      asyncio.wait(
       pending, 
       return_when=concurrent.futures.FIRST_COMPLETED 
      ) 
     ) 
     for job in done: 
      yield job.result() 

Diesen Ansatz, der eine Liste von pending Erhaltung von Arbeitsplätzen, hat den Vorteil, dass wir es anpassen können, um Aufträge auf der pending Liste im laufenden Betrieb hinzuzufügen. Dies ist nützlich in Anwendungsfällen, in denen unsere asynchronen Jobs der Warteschlange eine unvorhersehbare Anzahl weiterer Jobs hinzufügen können - wie eine Webspinne, die allen Links auf jeder Seite folgt, die sie besucht.