Python-asyncio-Modul

asyncioist ein Modul in der Python-Standardbibliothek, das zum Schreiben von asynchronem I/O-Operationscode verwendet wird.

asyncio bietet eine effiziente Möglichkeit, gleichzeitige Aufgaben zu bearbeiten, besonders geeignet für I/O-intensive Operationen wie Netzwerkanfragen, Dateilesen und -schreiben usw.

Indem Sieasyncio, können Sie mehrere Aufgaben gleichzeitig in einem einzigen Thread verarbeiten, ohne mehrere Threads oder Prozesse zu verwenden.

Warum brauchen wir asyncio?

In der traditionellen synchronen Programmierung blockiert das Programm, wenn eine Aufgabe auf den Abschluss einer I/O-Operation (wie z. B. einer Netzwerkanfrage) warten muss, bis die Operation abgeschlossen ist. Dies führt zu geringer Effizienz, insbesondere beim Umgang mit einer großen Anzahl von I/O-Operationen.

asyncioDurch die Einführung des asynchronen Programmiermodells kann das Programm während des Wartens auf I/O-Operationen mit der Ausführung anderer Aufgaben fortfahren, wodurch die Nebenläufigkeit und Effizienz des Programms verbessert werden.

Stellen Sie sich vor, Sie betreiben ein Restaurant:

  • Synchroner Modus (normale Funktionen):Sie haben nur einen Koch. Gast A bestellt ein Steak, und der Koch beginnt, das Steak zu braten (dies erfordert 5 Minuten Wartezeit). Während dieser 5 Minuten des Bratens ist der Koch vollständig beschäftigt und kann nichts anderes tun. Selbst wenn Gast B nur ein Glas Wasser möchte, muss er warten.
  • Asynchroner Modus (asyncio):Sie haben mehrere Köche (eigentlich immer noch einen, aber einen sehr intelligenten). Nachdem der Koch beginnt, das Steak von Gast A zu braten, merkt er, dass er warten muss. Er markiert das Steak sofort als wartend und dreht sich dann um, um Gast B Wasser einzugießen. Nachdem er das Wasser eingegossen hat, kommt er zurück, um zu prüfen, ob das Steak fast fertig ist. Wenn es noch nicht fertig ist, kann er sich um die Bestellung von Gast C kümmern. Auf diese Weise arbeitet der Koch (CPU) während der Wartezeit auf I/O (wie das Braten eines Steaks, Netzwerkanfragen, Lesen und Schreiben von Dateien) stets effizient.

asyncio ist die Standardbibliothek, die Python verwendet, um diesen intelligenten Arbeitsmodus zu implementieren. Sie ermöglicht es Ihnen, nebenläufigen Code mit einem einzelnen Thread zu schreiben, besonders geeignet für I/O-intensive Szenarien wie Web-Crawler, Webserver, Microservices usw.

Ihr Kern sind die Ereignisschleife, Koroutinen und Aufgaben.

Kernkonzepte von asyncio

1. Koroutine

Eine Koroutine istasyncioeines der Kernkonzepte. Sie ist eine spezielle Funktion, die während der Ausführung pausieren und später fortgesetzt werden kann. Koroutinen werden definiert mit demasync defSchlüsselwort und durch dasawaitSchlüsselwort, Ausführung anhalten, um auf den Abschluss einer asynchronen Operation zu warten.

Beispiel

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

2. Ereignisschleife

Die Ereignisschleife istasynciodie Kernkomponente, verantwortlich für die Planung und Ausführung von Coroutinen. Sie prüft kontinuierlich, ob Aufgaben ausgeführt werden müssen, und ruft nach Abschluss der Aufgaben die entsprechenden Callback-Funktionen auf.

Beispiel

async def main():
    await say_hello()

asyncio.run(main())

3. Aufgabe

Eine Aufgabe ist ein Wrapper um eine Coroutine und stellt eine Coroutine dar, die ausgeführt wird oder ausgeführt werden soll. Sie können dieasyncio.create_task()Funktion verwenden, um eine Aufgabe zu erstellen und sie zur Ereignisschleife hinzuzufügen.

Beispiel

async def main():
    task = asyncio.create_task(say_hello())
    await task

4. Future

Futureist ein Objekt, das das Ergebnis einer asynchronen Operation darstellt. Es wird normalerweise in Low-Level-APIs verwendet, um eine Operation darzustellen, die noch nicht abgeschlossen ist. Du kannst dasawaitSchlüsselwort zum Warten aufFutureum sie abzuschließen.

Beispiel

async def main():
    future = asyncio.Future()
    await future

Grundlegende Verwendung und Codebeispiele

Lass uns die oben genannten Konzepte anhand eines klassischen Beispiels für den gleichzeitigen Zugriff auf mehrere URLs verstehen.

Angenommen, wir müssen den Inhalt von drei verschiedenen URLs abrufen. Bei einem synchronen Ansatz werden Aufgaben sequenziell ausgeführt, und die Gesamtzeit ist die Summe der drei Anfragezeiten. Mitasyncio, können wir diese drei Anfragen gleichzeitig senden, und die Gesamtzeit liegt nahe an der langsamsten Anfrage.

Synchrone Version (zum Vergleich)

Beispiel

import time
import requests

def fetch_url(url):
    """Simuliere eine zeitaufwändige Netzwerkanfrage (synchrone Version)"""
    print(f"Abruf starten: {url}")
    time.sleep(2)  # Simuliere 2 Sekunden Netzwerkverzögerung
    print(f"Abruf abgeschlossen: {url}")
    return f"Daten von {url}"

def main_sync():
    urls = ['https://example.com/1', 'https://example.com/2', 'https://example.com/3']
    results = []
    start = time.time()
   
    for url in urls:
        result = fetch_url(url)  # Muss warten, bis der vorherige abgeschlossen ist, bevor der nächste startet
        results.append(result)
   
    end = time.time()
    print(f"Gesamtzeit der synchronen Version: {end - start:.2f} Sekunden")
    print(f"Ergebnisse: {results}")

if __name__ == "__main__":
    main_sync()

Erwartete Ausgabe：

开始获取: https://example.com/1
完成获取: https://example.com/1
开始获取: https://example.com/2
完成获取: https://example.com/2
开始获取: https://example.com/3
完成获取: https://example.com/3
同步版本总耗时: 6.00 秒
结果: [‘来自 https://example.com/1 的数据‘, ‘来自 https://example.com/2 的数据‘, ‘来自 https://example.com/2 的数据‘]

Es dauert insgesamt etwa 6 Sekunden.

Asynchrone Version (mit asyncio)

Wir müssenaiohttpBibliothek zum Ersetzenrequestsfür asynchrone HTTP-Anfragen. Zuerst installieren Sie sie:pip install aiohttp。

Beispiel

import asyncio
import aiohttp
import time

async def fetch_url_async(session, url):
    """Simuliert eine zeitaufwändige Netzwerkanfrage (asynchrone Version)"""
    print(f"Asynchrones Abrufen starten: {url}")
    # Hinweis: Hier verwenden wir die asynchrone get-Methode von aiohttp und warten mit await
    async with session.get(url) as response:
        # Simuliere, dass die Verarbeitung der Antwort ebenfalls Zeit benötigt
        await asyncio.sleep(2)  # Verwende asyncio.sleep, um das Warten auf I/O zu simulieren; es blockiert den Thread nicht
        text = await response.text()
        print(f"Asynchrones Abrufen abgeschlossen: {url}")
        return f"Daten von {url} (Länge: {len(text)})"

async def main_async():
    urls = ['https://httpbin.org/get', 'https://httpbin.org/delay/1', 'https://httpbin.org/headers']
   
    async with aiohttp.ClientSession() as session:  # Erstelle eine asynchrone HTTP-Session
        # Erstelle eine Aufgabe (Task) für jede URL
        tasks = []
        for url in urls:
            # create_task fügt die Coroutine zur Ereignisschleife hinzu und beginnt sofort mit der Planung
            task = asyncio.create_task(fetch_url_async(session, url))
            tasks.append(task)
       
        print("Alle Aufgaben erstellt, starte gleichzeitige Ausführung...")
       
        # Verwende asyncio.gather, um alle Aufgaben gleichzeitig auszuführen und zu warten, bis alle abgeschlossen sind
        # gather gibt eine Liste von Ergebnissen zurück, deren Reihenfolge mit der Reihenfolge der übergebenen Aufgaben übereinstimmt
        results = await asyncio.gather(*tasks)
       
        return results

if __name__ == "__main__":
    start = time.time()
    # asyncio.run() ist eine bequeme Möglichkeit, die Ereignisschleife zu starten und die Coroutine der obersten Ebene auszuführen
    final_results = asyncio.run(main_async())
    end = time.time()
   
    print(f"\n"Asynchrone Version Gesamtzeit: {end - start:.2f} Sekunden")
    for res in final_results:
        print(res)

Erwartete Ausgabe：

所有任务已创建，开始并发执行...
开始异步获取: https://httpbin.org/get
开始异步获取: https://httpbin.org/delay/1
开始异步获取: https://httpbin.org/headers
（大约 2 秒后，所有请求几乎同时完成）
完成异步获取: https://httpbin.org/headers
完成异步获取: https://httpbin.org/get
完成异步获取: https://httpbin.org/delay/1

异步版本总耗时: 2.10 秒  # 注意！总耗时远小于 6 秒
来自 https://httpbin.org/get 的数据 (长度: 274)
来自 https://httpbin.org/delay/1 的数据 (长度: 392)
来自 https://httpbin.org/headers 的数据 (长度: 177)

Codeanalyse：

  • async def: definiert die Coroutine-Funktionfetch_url_asyncundmain_async。
  • await: infetch_url_async, wirawait session.get()undawait response.text(), dies teilt dem Event-Loop mit: "Diese Netzwerkanfrage dauert; führe zuerst andere bereitstehende Aufgaben aus."
  • asyncio.create_task(): verpackt diefetch_url_asyncCoroutine in eineTask, wodurch sie vom Event-Loop eingeplant werden kann und Nebenläufigkeit erreicht wird.
  • asyncio.gather(*tasks): eine sehr praktische Funktion, die alle übergebenen Coroutinen/Tasks nebenläufig ausführt, wartet, bis alle abgeschlossen sind, und schließlich alle Ergebnisse einsammelt.
  • asyncio.run(main_async()): der empfohlene Ansatz in Python 3.7+, verantwortlich für das Erstellen des Event-Loops, das Ausführen der Coroutine und das Schließen der Schleife.

Wichtige Funktionen und Parameterbeschreibungen

Die folgende Tabelle listetasyncioeinige der am häufigsten verwendeten High-Level-Funktionen:

Funktion Hauptzweck Allgemeine Parameterbeschreibungen
asyncio.run(coro, *, debug=False) Führt eine Top-Level-Coroutine aus und verwaltet den Lebenszyklus des Event-Loops. Es ist der Haupteinstiegspunkt des Programms. coro: das auszuführende Coroutine-Objekt.
debug: gesetzt aufTrueum den Debug-Modus der Event-Loop zu aktivieren.
asyncio.create_task(coro, *, name=None) Verpackt eine Coroutine in einTask-Objekt und reiht sie zur Planung in die Event-Loop ein. Dies ist der wichtigste Weg, um Nebenläufigkeit zu erreichen. coro: das Coroutine-Objekt, das verpackt werden soll.
name: (Python 3.8+) Name für den Task angeben, um das Debugging zu erleichtern.
asyncio.gather(*aws, return_exceptions=False) Führt gleichzeitigmehrere asynchrone Tasks aus (awskann Coroutinen, Tasks usw. annehmen), wartet auf den Abschluss aller und gibt eine Liste von Ergebnissen zurück. *aws: variable Argumente; übergeben Sie mehrere asynchrone Objekte.
return_exceptions: ist standardmäßigFalse, jede Ausnahme, die von einem Task ausgelöst wird, wird sofort angatherden Aufrufer. Setzen Sie aufTruewenn gesetzt, werden Ausnahmen als Teil der Ergebnisse zurückgegeben.
asyncio.sleep(delay, result=None) Asynchronschläft für die angegebene Anzahl von Sekunden. Dies ist der entscheidende Unterschied zutime.sleep(blockierend). delay: Anzahl der Sekunden zum Schlafen.
result: Der Wert, der nach dem Ende der Wartezeit zurückgegeben wird.
asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED) Führt Aufgaben nebenläufig aus und wartet, bis die angegebene Bedingung erfüllt ist. Gibt zwei Mengen zurück(done, pending), die die abgeschlossenen bzw. nicht abgeschlossenen Aufgaben sind. aws: Sammlung asynchroner Objekte.
timeout: Timeout-Dauer (Sekunden).
return_when: Bestimmt, wann zurückgegeben wird; Optionen:FIRST_COMPLETED(erste abgeschlossen),FIRST_EXCEPTION(erste Ausnahme),ALL_COMPLETED(alle abgeschlossen, Standard).
asyncio.to_thread(func, /, *args, **kwargs) (Python 3.9+) Führt eine gewöhnliche, potenziell blockierende synchrone Funktion in einem separaten Thread aus und gibt einawaitawaitbare Coroutine. Wird für die Behandlung CPU-intensiver oder blockierender I/O verwendet. func: Die synchrone Funktion, die im Thread ausgeführt werden soll.
*args, **kwargs: Argumente, die an die Funktion übergeben werden.

Visuelles Verständnis: Ablauf der asynchronen Aufgabenplanung

Diagrammerklärung: Dieses Flussdiagramm zeigt, wie die Ereignisschleife wie ein Dispatcher funktioniert. Sie verwaltet eine Aufgabenwarteschlange. Wenn eine Aufgabe an einen Punkt gelangt, an dem sieawait(z. B. auf eine Netzwerkantwort wartet), wird sie angehalten. Die Ereignisschleife findet sofort die nächste ausführbare (bereite) Aufgabe aus der Warteschlange und führt sie aus. Wenn der I/O-Vorgang der angehaltenen Aufgabe abgeschlossen ist, erhält die Ereignisschleife eine Benachrichtigung, ändert den Status der Aufgabe zurück auf bereit und setzt ihre Ausführung zu einem späteren Zeitpunkt fort. Auf diese Weise wird die CPU während I/O-Wartezeiten voll ausgelastet, um andere Aufgaben auszuführen, wodurch Nebenläufigkeit innerhalb eines einzelnen Threads erreicht wird.

Grundlegende Verwendung von asyncio

1. Eine Coroutine ausführen

Um eine Coroutine auszuführen, können Sie dieasyncio.run()Funktion verwenden. Sie erstellt eine Ereignisschleife und führt die angegebene Coroutine aus.

Beispiel

import asyncio

async def main():
    print("Start")
    await asyncio.sleep(1)
    print("End")

asyncio.run(main())

2. Mehrere Aufgaben gleichzeitig ausführen

Sie können dieasyncio.gather()Funktion, um mehrere Coroutinen gleichzeitig auszuführen und zu warten, bis alle abgeschlossen sind.

Beispiel

import asyncio

async def task1():
    print("Aufgabe 1 gestartet")
    await asyncio.sleep(1)
    print("Aufgabe 1 beendet")

async def task2():
    print("Aufgabe 2 gestartet")
    await asyncio.sleep(2)
    print("Aufgabe 2 beendet")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

3. Timeout-Steuerung

Sie können dieasyncio.wait_for()Funktion, um ein Timeout für eine Coroutine festzulegen. Wenn die Coroutine nicht innerhalb der angegebenen Zeit abgeschlossen wird, wird eineasyncio.TimeoutErrorAusnahme ausgelöst.

Beispiel

import asyncio

async def long_task():
    await asyncio.sleep(10)
    print("Task beendet")

async def main():
    try:
        await asyncio.wait_for(long_task(), timeout=5)
    except asyncio.TimeoutError:
        print("Zeitüberschreitung der Aufgabe")

asyncio.run(main())

Anwendungsszenarien von asyncio

asyncioBesonders geeignet für die folgenden Szenarien:

  1. Netzwerkanfragen: wie HTTP-Anfragen, WebSocket-Kommunikation usw.
  2. Datei-I/O: wie asynchrones Lesen und Schreiben von Dateien.
  3. Datenbankoperationen: wie asynchroner Datenbankzugriff.
  4. Echtzeit-Datenverarbeitung: wie Echtzeit-Verarbeitung von Nachrichtenwarteschlangen.

Gängige Klassen, Methoden und Funktionen

1. Kernfunktionen

Methode/Funktion Beschreibung Beispiel
asyncio.run(coro) Die asynchrone Hauptfunktion ausführen (Python 3.7+) asyncio.run(main())
asyncio.create_task(coro) Eine Task erstellen und zur Ereignisschleife hinzufügen task = asyncio.create_task(fetch_data())
asyncio.gather(*coros) Mehrere Coroutinen gleichzeitig ausführen await asyncio.gather(task1, task2)
asyncio.sleep(delay) Asynchrones Warten (nicht blockierend) await asyncio.sleep(1)
asyncio.wait(coros) Steuern, wie Tasks abgeschlossen werden done, pending = await asyncio.wait([task1, task2])

2. Ereignisschleife

Methode Beschreibung Beispiel
loop.run_until_complete(future) Ausführen, bis die Task abgeschlossen ist loop.run_until_complete(main())
loop.run_forever() Die Ereignisschleife endlos ausführen loop.run_forever()
loop.stop() Die Ereignisschleife stoppen loop.stop()
loop.close() Die Ereignisschleife schließen loop.close()
loop.call_soon(callback) Eine Callback-Funktion zur sofortigen Ausführung einplanen loop.call_soon(print, "Hello")
loop.call_later(delay, callback) Einen Callback mit einer Verzögerung ausführen loop.call_later(5, callback)

3. Koroutinen und Tasks

Methode/Dekorator Beschreibung Beispiel
@asyncio.coroutine Koroutine-Dekorator (veraltet, Python 3.4-3.7) @asyncio.coroutine
def old_coro():
async def Eine Koroutine definieren (Python 3.5+) async def fetch():
task.cancel() Eine Task abbrechen task.cancel()
task.done() Prüfen, ob eine Task abgeschlossen ist if task.done():
task.result() Das Ergebnis der Task abrufen (erfordert, dass die Task abgeschlossen ist) data = task.result()

4. Synchronisationsprimitive (ähnlich wiethreading）

Klasse Beschreibung Beispiel
asyncio.Lock() Asynchroner Mutex-Lock lock = asyncio.Lock()
async with lock:
asyncio.Event() Event-Benachrichtigung event = asyncio.Event()
await event.wait()
asyncio.Queue() Asynchrone Warteschlange queue = asyncio.Queue()
await queue.put(item)
asyncio.Semaphore() Semaphore sem = asyncio.Semaphore(5)
async with sem:

5. Netzwerk und Subprozesse

Methode/Klasse Beschreibung Beispiel
asyncio.open_connection() Eine TCP-Verbindung herstellen reader, writer = await asyncio.open_connection('host', 80)
asyncio.start_server() Einen TCP-Server erstellen server = await asyncio.start_server(handle, '0.0.0.0', 8888)
asyncio.create_subprocess_exec() Einen Subprozess erstellen proc = await asyncio.create_subprocess_exec('ls')

6. Hilfsprogramme

Methode Beschreibung Beispiel
asyncio.current_task() Die aktuelle Aufgabe abrufen task = asyncio.current_task()
asyncio.all_tasks() Alle Aufgaben abrufen tasks = asyncio.all_tasks()
asyncio.shield(coro) Eine Aufgabe vor Abbruch schützen await asyncio.shield(critical_task)
asyncio.wait_for(coro, timeout) Mit einem Timeout warten try: await asyncio.wait_for(task, 5)

Beispiel

1. Einfaches Koroutinen-Beispiel

Beispiel

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(hello())  # Python 3.7+

2. Aufgaben nebenläufig ausführen

Beispiel

async def fetch(url):
    print(f"Rufe {url} ab")
    await asyncio.sleep(2)
    return f"Daten von {url}"

async def main():
    results = await asyncio.gather(
        fetch("url1.com"),
        fetch("url2.com")
    )
    print(results)

asyncio.run(main())

3. Eine asynchrone Warteschlange verwenden

Beispiel

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        await asyncio.sleep(0.1)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Konsumiert {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

Hinweise

  1. Python-Version: Einige Funktionen erfordern Python 3.7+ (wie zum Beispielasyncio.run()）。

  2. Blockierende Operationen: Vermeiden Sie synchronen blockierenden Code in Coroutinen (wie zum Beispieltime.sleep()）。

  3. Debugging: Setzen Sie diePYTHONASYNCIODEBUG=1Umgebungsvariable, um den Debug-Modus zu aktivieren.

  4. Aufgaben abbrechen: Eine abgebrochene Aufgabe wirftCancelledError, was ordnungsgemäß behandelt werden muss.

Weitere Erweiterungen