Python Language
Asyncio-Modul

Coroutine und Delegation Syntax

Vor der Veröffentlichung von Python 3.5+ verwendete das asyncio Modul Generatoren, um asynchrone Aufrufe nachzuahmen, und verfügte daher über eine andere Syntax als das aktuelle Python 3.5-Release.

import asyncio

async def main():
    print(await func())

async def func():
    # Do time intensive stuff...
    return "Hello, world!"

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

import asyncio

@asyncio.coroutine
def main():
    print((yield from func()))

@asyncio.coroutine
def func():
    # Do time intensive stuff..
    return "Hello, world!"

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

import asyncio

async def cor1():
    print("cor1 start")
    for i in range(10):
        await asyncio.sleep(1.5)
        print("cor1", i)

async def cor2():
    print("cor2 start")
    for i in range(15):
        await asyncio.sleep(1)
        print("cor2", i)

loop = asyncio.get_event_loop()
cors = asyncio.wait([cor1(), cor2()])
loop.run_until_complete(cors)

Asynchrone Executoren

Hinweis: Verwendet die Python 3.5+ async / await-Syntax

asyncio unterstützt die Verwendung von in concurrent.futures gefundenen Executor Objekten, um Aufgaben asynchron zu planen. Ereignisschleifen haben die Funktion run_in_executor() die ein Executor Objekt, ein Callable und die Callable-Parameter übernimmt.

Planen einer Aufgabe für einen Executor

import asyncio
from concurrent.futures import ThreadPoolExecutor

def func(a, b):
    # Do time intensive stuff...
    return a + b

async def main(loop):
    executor = ThreadPoolExecutor()
    result = await loop.run_in_executor(executor, func, "Hello,", " world!")
    print(result)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))

Jede Ereignisschleife verfügt außerdem über einen "Standard" Executor Slot, der einem Executor zugewiesen werden kann. Um einen Executor zuzuweisen und Aufgaben aus der Schleife zu planen, verwenden Sie die Methode set_default_executor() .

import asyncio
from concurrent.futures import ThreadPoolExecutor

def func(a, b):
    # Do time intensive stuff...
    return a + b

async def main(loop):
    # NOTE: Using `None` as the first parameter designates the `default` Executor.
    result = await loop.run_in_executor(None, func, "Hello,", " world!")
    print(result)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.set_default_executor(ThreadPoolExecutor())
    loop.run_until_complete(main(loop))

Es gibt zwei Haupttypen von Executor in concurrent.futures , den ThreadPoolExecutor und den ProcessPoolExecutor . Der ThreadPoolExecutor enthält einen Pool von Threads , die entweder manuell auf eine bestimmte Anzahl von Fäden durch den Konstruktor oder Standardwerte für die Anzahl der Kerne auf dem Maschinenzeit 5. eingestellt werden ThreadPoolExecutor verwendet den Pool von Threads Aufgaben es auszuführen zugeordnet ist und Im Allgemeinen besser bei CPU-gebundenen Operationen als bei E / A-gebundenen Operationen. ProcessPoolExecutor dies mit dem ProcessPoolExecutor der für jede ihm zugewiesene Aufgabe einen neuen Prozess erzeugt. Der ProcessPoolExecutor kann nur Aufgaben und Parameter übernehmen, die kommissionierbar sind. Die häufigsten nicht auswählbaren Aufgaben sind die Methoden von Objekten. Wenn Sie die Methode eines Objekts als Task in einem Executor ThreadPoolExecutor müssen, müssen Sie einen ThreadPoolExecutor .

UVLoop verwenden

uvloop ist eine Implementierung für asyncio.AbstractEventLoop basierend auf libuv (Wird von nodejs verwendet). Es ist mit 99% der asyncio Funktionen asyncio und ist viel schneller als das herkömmliche asyncio.EventLoop . uvloop ist derzeit unter Windows nicht verfügbar. Installieren Sie es mit pip install uvloop .

import asyncio
import uvloop

if __name__ == "__main__":
    asyncio.set_event_loop(uvloop.new_event_loop())
    # Do your stuff here ...

Sie können die Ereignisschleifen-Factory auch ändern, indem Sie die EventLoopPolicy auf die in uvloop .

import asyncio
import uvloop

if __name__ == "__main__":
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = asyncio.new_event_loop()

Synchronisationsprimitiv: Ereignis

Konzept

Verwenden Sie ein Event , um die Planung mehrerer Coroutinen zu synchronisieren .

Vereinfacht gesagt, ist ein Ereignis wie eine Waffe, die auf ein Rennen abgefeuert wird: Es lässt die Läufer die Startblöcke verlassen.

Beispiel

import asyncio

# event trigger function
def trigger(event):
    print('EVENT SET')
    event.set()  # wake up coroutines waiting

# event consumers
async def consumer_a(event):
    consumer_name = 'Consumer A'
    print('{} waiting'.format(consumer_name))
    await event.wait()
    print('{} triggered'.format(consumer_name))

async def consumer_b(event):
    consumer_name = 'Consumer B'
    print('{} waiting'.format(consumer_name))
    await event.wait()
    print('{} triggered'.format(consumer_name))

# event
event = asyncio.Event()

# wrap coroutines in one future
main_future = asyncio.wait([consumer_a(event),
                            consumer_b(event)])

# event loop
event_loop = asyncio.get_event_loop()
event_loop.call_later(0.1, functools.partial(trigger, event))  # trigger event in 0.1 sec

# complete main_future
done, pending = event_loop.run_until_complete(main_future)

Ausgabe:

Verbraucher B wartet
Verbraucher Ein Warten
EVENT SET
Verbraucher B ausgelöst
Consumer A ausgelöst

Ein einfacher Websocket

Hier asyncio wir mit asyncio einen einfachen Echo- asyncio . Wir definieren Coroutinen für die Verbindung zu einem Server und zum Senden / Empfangen von Nachrichten. Die Communcations des websocket werden in einem laufen main Koroutine, die durch eine Ereignisschleife ausgeführt wird. Dieses Beispiel wurde aus einem früheren Beitrag geändert.

import asyncio
import aiohttp

session = aiohttp.ClientSession()                          # handles the context manager
class EchoWebsocket:
    
    async def connect(self):
        self.websocket = await session.ws_connect("wss://echo.websocket.org")
        
    async def send(self, message):
        self.websocket.send_str(message)

    async def receive(self):
        result = (await self.websocket.receive())
        return result.data

async def main():
    echo = EchoWebsocket()
    await echo.connect()
    await echo.send("Hello World!")
    print(await echo.receive())                            # "Hello World!"


if __name__ == '__main__':
    # The main loop
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Häufiges Missverständnis über Asyncio

Das häufigste Missverständnis über asnycio besteht asnycio darin, dass Sie alle Aufgaben parallel ausführen können, indem Sie die GIL (globale Interpretersperre) umgehen und parallel dazu blockierende Jobs (in separaten Threads) ausführen. es tut nicht

asyncio (und Bibliotheken, die mit asyncio ) bauen auf Coroutines auf: Funktionen, die (gemeinsam) den Steuerungsfluss an die aufrufende Funktion zurückgeben. Beachten Sie asyncio.sleep in den obigen Beispielen. Dies ist ein Beispiel für eine nicht blockierende Coroutine, die 'im Hintergrund' wartet und den Steuerungsfluss an die aufrufende Funktion zurückgibt (wenn mit await aufgerufen). time.sleep ist ein Beispiel für eine time.sleep . Der Ablauf des Programms wird dort einfach time.sleep und erst nach time.sleep .

Ein reales Beispiel ist die requests die (vorerst) nur aus Sperrfunktionen besteht. Es gibt keine Parallelität, wenn Sie eine seiner Funktionen innerhalb von asyncio . aiohttp dagegen wurde für asyncio . Seine Coroutinen werden gleichzeitig laufen.

• Wenn Sie langlaufende CPU-gebundene Aufgaben haben, die Sie parallel asyncio ist asyncio nichts für Sie. dazu benötigen Sie threads oder multiprocessing .

• wenn Sie IO-gebundene Arbeitsplätze ausgeführt wird , können Sie sie gleichzeitig mit laufen asyncio .