Python Language
Módulo asyncio

Sintaxis de Coroutine y Delegación

Antes de que se lanzara Python asyncio , el módulo asyncio usaba generadores para imitar las llamadas asíncronas y, por lo tanto, tenía una sintaxis diferente a la versión actual de 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)

Ejecutores asincronos

Nota: utiliza la sintaxis de Python 3.5+ async / await

asyncio admite el uso de objetos Executor que se encuentran en concurrent.futures para programar tareas de forma asíncrona. Los bucles de eventos tienen la función run_in_executor() que toma un objeto Executor , un Callable y los parámetros del Callable.

Programando una tarea para 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))

Cada ciclo de eventos también tiene una ranura de Executor "predeterminada" que se puede asignar a un Executor . Para asignar un Executor y programar tareas desde el bucle, utilice el método 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))

Hay dos tipos principales de Executor en concurrent.futures , ThreadPoolExecutor y ProcessPoolExecutor . ThreadPoolExecutor contiene un conjunto de subprocesos que se pueden establecer manualmente en un número específico de subprocesos a través del constructor o por defecto el número de núcleos en los tiempos de la máquina 5. El ThreadPoolExecutor utiliza el conjunto de subprocesos para ejecutar tareas asignadas a él En general, es mejor en operaciones vinculadas a la CPU en lugar de operaciones vinculadas de E / S Contraste eso con el ProcessPoolExecutor que genera un nuevo proceso para cada tarea asignada. ProcessPoolExecutor solo puede tomar tareas y parámetros que son seleccionables. Las tareas no recolectables más comunes son los métodos de los objetos. Si debe programar el método de un objeto como una tarea en un Executor , debe usar un ThreadPoolExecutor .

Usando UVLoop

uvloop es una implementación para asyncio.AbstractEventLoop basada en libuv (utilizada por nodejs). Cumple con el 99% de asyncio funciones de asyncio y es mucho más rápido que el asyncio.EventLoop tradicional. asyncio.EventLoop . uvloop actualmente no está disponible en Windows, instálelo con pip install uvloop .

import asyncio
import uvloop

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

También se puede cambiar la fábrica de bucles de eventos configurando EventLoopPolicy a la de uvloop .

import asyncio
import uvloop

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

Primitiva de sincronización: Evento

Concepto

Utilice un Event para sincronizar la programación de múltiples coroutines .

En pocas palabras, un evento es como el disparo en una carrera: permite que los corredores salgan de los bloques de salida.

Ejemplo

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)

Salida:

Consumidor B en espera
Consumidor A esperando
SET DE EVENTOS
Consumidor B activado
Consumidor A activado

Un simple websocket

Aquí hacemos un simple websocket eco utilizando asyncio . Definimos las rutinas para conectarse a un servidor y enviar / recibir mensajes. Las comunicaciones del websocket se ejecutan en una rutina main , que se ejecuta mediante un bucle de eventos. Este ejemplo es modificado de una publicación anterior .

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

Error común sobre asyncio

Probablemente, la idea errónea más común acerca de asnycio es que le permite ejecutar cualquier tarea en paralelo: eludir el GIL (bloqueo global del intérprete) y, por lo tanto, ejecutar trabajos de bloqueo en paralelo (en subprocesos separados). no lo hace!

asyncio (y las bibliotecas creadas para colaborar con asyncio ) se basan en coroutines: funciones que (en colaboración) devuelven el flujo de control a la función de llamada. asyncio.sleep en cuenta asyncio.sleep en los ejemplos anteriores. este es un ejemplo de una coroutina no bloqueante que espera 'en el fondo' y devuelve el flujo de control a la función de llamada (cuando se llama con await ). time.sleep es un ejemplo de una función de bloqueo. el flujo de ejecución del programa se detendrá allí y solo regresará después de que time.sleep haya terminado.

un ejemplo real es la biblioteca de requests que consiste (por el momento) solo en funciones de bloqueo. no hay concurrencia si llama a cualquiera de sus funciones dentro de asyncio . aiohttp por otro lado fue construido con asyncio en mente. sus coroutines correrán concurrentemente.

• Si tiene tareas vinculadas a la CPU de larga ejecución que le gustaría ejecutar en paralelo, asyncio no es para usted. Para eso necesitas threads o multiprocessing .

• Si tiene trabajos en ejecución enlazados a IO, puede ejecutarlos simultáneamente usando asyncio .