Python Language
Modulo Asyncio

Sintassi di Coroutine e Delegazione

Prima che Python 3.5+ venisse rilasciato, il modulo asyncio usava i generatori per imitare le chiamate asincrone e quindi aveva una sintassi diversa rispetto all'attuale versione di Python 3.5.

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)

Esecutori asincroni

Nota: utilizza Python 3.5+ async / attende la sintassi

asyncio supporta l'uso di oggetti Executor trovati in concurrent.futures per la pianificazione delle attività in modo asincrono. I loop di eventi hanno la funzione run_in_executor() che accetta un oggetto Executor , un Callable e i parametri di Callable.

Pianificazione di un'attività per un 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))

Ogni loop di eventi ha anche uno slot Executor "predefinito" che può essere assegnato a un Executor . Per assegnare un Executor e pianificare le attività dal ciclo, utilizzare il metodo 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))

Esistono due tipi principali di Executor in concurrent.futures , ThreadPoolExecutor e ProcessPoolExecutor . ThreadPoolExecutor contiene un pool di thread che può essere impostato manualmente su un numero specifico di thread tramite il costruttore o predefinito sul numero di core sulla macchina time 5. ThreadPoolExecutor utilizza il pool di thread per eseguire attività assegnate ad esso ed è generalmente meglio alle operazioni associate alla CPU piuttosto che alle operazioni associate all'I / O. Contrasto a ProcessPoolExecutor che genera un nuovo processo per ogni attività assegnata ad esso. ProcessPoolExecutor può solo eseguire attività e parametri che sono selezionabili. I compiti non selezionabili più comuni sono i metodi degli oggetti. Se è necessario pianificare il metodo di un oggetto come attività in un Executor è necessario utilizzare ThreadPoolExecutor .

Utilizzando UVLoop

uvloop è un'implementazione per asyncio.AbstractEventLoop basato su libuv (usato da nodejs). È conforme al 99% delle funzionalità di asyncio ed è molto più veloce rispetto al tradizionale asyncio.EventLoop . uvloop è attualmente disponibile su Windows, installalo con pip install uvloop .

import asyncio
import uvloop

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

Si può anche cambiare il factory del loop degli eventi impostando EventLoopPolicy su quello in uvloop .

import asyncio
import uvloop

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

Sincronizzazione primitiva: evento

Concetto

Utilizzare un Event per sincronizzare la pianificazione di più coroutine .

In parole povere, un evento è come la pistola sparata a una corsa in corsa: lascia i corridori fuori dai blocchi di partenza.

Esempio

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)

Produzione:

Consumatore B in attesa
Consumatore A in attesa
SET EVENTO
Consumer B attivato
Consumatore A attivato

Un Websocket semplice

Qui creiamo un semplice web echo usando asyncio . Definiamo le coroutine per la connessione a un server e l'invio / ricezione di messaggi. Le comunicazioni del websocket vengono eseguite in una coroutine main , che viene eseguita da un ciclo di eventi. Questo esempio è stato modificato da un post precedente .

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

Idee sbagliate comuni sull'asyncio

probabilmente il più comune malinteso su asnycio è che ti permette di eseguire qualsiasi attività in parallelo - asnycio il GIL (global interpreter lock) e quindi eseguire processi di blocco in parallelo (su thread separati). non è così!

asyncio (e le librerie che sono costruite per collaborare con asyncio ) si basano su coroutine: funzioni che (in modo collaborativo) restituiscono il flusso di controllo alla funzione chiamante. nota asyncio.sleep negli esempi sopra. questo è un esempio di una coroutine non bloccante che attende "in background" e restituisce il flusso di controllo alla funzione chiamante (quando chiamata con await ). time.sleep è un esempio di una funzione di blocco. il flusso di esecuzione del programma si fermerà qui e ritorna solo dopo che il time.sleep è time.sleep .

un esempio reale è la libreria delle requests che consiste (per il momento) solo sulle funzioni di blocco. non c'è concorrenza se si chiama una delle sue funzioni all'interno di asyncio . aiohttp d'altra parte, è stato costruito asyncio ad asyncio . le sue coroutine verranno eseguite contemporaneamente.

• se hai attività con CPU a lungo termine che ti piacerebbe eseguire in parallelo, asyncio non fa per te. per questo hai bisogno di threads o multiprocessing .

• se sono in esecuzione lavori con asyncio IO, è possibile eseguirli contemporaneamente utilizzando asyncio .