Python Language
Asyncio-module

Coroutine- en delegatiesyntaxis

Voordat Python 3.5+ werd uitgebracht, gebruikte de asyncio module generators om asynchrone oproepen na te bootsen en had dus een andere syntaxis dan de huidige 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 uitvoerders

Opmerking: maakt gebruik van de Python 3.5+ async / wacht op syntaxis

asyncio ondersteunt het gebruik van Executor objecten gevonden in concurrent.futures voor het asynchroon plannen van taken. run_in_executor() hebben de functie run_in_executor() die een Executor object, een Callable en de parameters van Callable heeft.

Een taak plannen voor een 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))

Elke eventlus heeft ook een "standaard" Executor slot dat kan worden toegewezen aan een Executor . Om een toewijzen Executor en taken plannen van de lus u gebruik set_default_executor() methode.

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))

Er zijn twee hoofdtypen Executor in concurrent.futures , de ThreadPoolExecutor en de ProcessPoolExecutor . De ThreadPoolExecutor bevat een pool met threads die handmatig kunnen worden ingesteld op een specifiek aantal threads via de constructor of standaard ingesteld op het aantal cores op de machinetijd 5. De ThreadPoolExecutor gebruikt de pool met threads om toegewezen taken uit te voeren en is over het algemeen beter bij CPU-gebonden bewerkingen in plaats van I / O-gebonden bewerkingen. Vergelijk dat met de ProcessPoolExecutor die een nieuw proces voortbrengt voor elke taak die eraan is toegewezen. De ProcessPoolExecutor kan alleen taken en parameters aannemen die kunnen worden geselecteerd. De meest voorkomende niet-kiesbare taken zijn de methoden van objecten. Als u de methode van een object als een taak in een Executor moet plannen, moet u een ThreadPoolExecutor .

UVLoop gebruiken

uvloop is een implementatie voor de asyncio.AbstractEventLoop gebaseerd op libuv (gebruikt door nodejs). Het is compatibel met 99% van de asyncio functies en is veel sneller dan de traditionele asyncio.EventLoop . uvloop is momenteel niet beschikbaar op Windows, installeer het met pip install uvloop .

import asyncio
import uvloop

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

Je kunt de event- EventLoopPolicy ook wijzigen door EventLoopPolicy te stellen op degene in uvloop .

import asyncio
import uvloop

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

Synchronisatie primitief: gebeurtenis

Concept

Gebruik een Event om de planning van meerdere coroutines te synchroniseren .

Simpel gezegd, een evenement is als een schietpartij op een lopende race: het laat de lopers de startblokken verlaten.

Voorbeeld

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)

Output:

Consument B wacht
Consument Een wachten
GEBEURTENIS INSTELLEN
Consument B geactiveerd
Consument A geactiveerd

Een eenvoudige websocket

Hier maken we een eenvoudige echo-websocket met asyncio . We definiëren coroutines voor het verbinden met een server en het verzenden / ontvangen van berichten. De communcations van de WebSocket worden uitgevoerd in een main coroutine, dat wordt beheerd door een gebeurtenissenlus. Dit voorbeeld is gewijzigd van een eerdere post .

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())

Veel voorkomende misvatting over asyncio

waarschijnlijk is de meest voorkomende misvatting over asnycio dat je elke taak parallel kunt uitvoeren - de GIL omzeilen (global interpreter lock) en daarom blokkeertaken parallel uitvoeren (op afzonderlijke threads). dat doet het niet !

asyncio (en bibliotheken die zijn gebouwd om samen te werken met asyncio ) bouwen op coroutines: functies die (samen) de besturingsstroom terugbrengen naar de aanroepfunctie. noteer asyncio.sleep in de bovenstaande voorbeelden. dit is een voorbeeld van een niet-blokkerende coroutine die 'op de achtergrond' wacht en de besturingsstroom terugvoert naar de aanroepfunctie (indien opgeroepen met await ). time.sleep is een voorbeeld van een blokkeerfunctie. de uitvoering van het programma stopt daar gewoon en keert pas terug nadat time.sleep is afgelopen.

een echt voorbeeld is de bibliotheek met requests die (voorlopig) alleen bestaat uit blokkeerfuncties. er is geen gelijktijdigheid als u een van zijn functies binnen asyncio . aiohttp daarentegen werd gebouwd met asyncio in gedachten. de coroutines zullen gelijktijdig worden uitgevoerd.

• als u langlopende CPU-gebonden taken hebt die u parallel wilt uitvoeren, asyncio is niets voor u. daarvoor heb je threads of multiprocessing .

• als u IO-gebonden taken uitvoert, kunt u deze tegelijkertijd uitvoeren met asyncio .