Python Language
Module Asyncio

Coroutine et syntaxe de délégation

Avant la sortie de Python 3.5+, le module asyncio utilisait des générateurs pour imiter les appels asynchrones et avait donc une syntaxe différente de la version actuelle 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)

Exécuteurs Asynchrones

Remarque: Utilise la syntaxe asynchrone / d'attente Python 3.5+

asyncio prend en charge l'utilisation des objets Executor trouvés dans concurrent.futures pour planifier des tâches de manière asynchrone. Les boucles d'événement ont la fonction run_in_executor() qui prend un objet Executor , un Callable et les paramètres de Callable.

Planification d'une tâche pour 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))

Chaque boucle d'événement dispose également d'un emplacement d' Executor "par défaut" pouvant être attribué à un Executor . Pour assigner un Executor et planifier des tâches depuis la boucle, vous utilisez la méthode 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))

Il existe deux principaux types d' Executor dans concurrent.futures , ThreadPoolExecutor et ProcessPoolExecutor . ThreadPoolExecutor contient un pool de threads pouvant être défini manuellement sur un nombre spécifique de threads via le constructeur ou par défaut sur le nombre de cœurs de la machine. ThreadPoolExecutor utilise le pool de threads pour exécuter les tâches qui lui sont affectées. généralement meilleure pour les opérations liées au processeur que pour les opérations liées aux E / S. Cela contraste avec ProcessPoolExecutor qui génère un nouveau processus pour chaque tâche qui lui est affectée. ProcessPoolExecutor ne peut que prendre en charge les tâches et les paramètres pouvant être capturés. Les tâches non-picklables les plus courantes sont les méthodes des objets. Si vous devez planifier la méthode d'un objet en tant que tâche dans un Executor vous devez utiliser un ThreadPoolExecutor .

Utiliser UVLoop

uvloop est une implémentation de asyncio.AbstractEventLoop basée sur libuv (Utilisé par nodejs). Il est compatible avec 99% des fonctionnalités asyncio et est beaucoup plus rapide que le traditionnel asyncio.EventLoop . uvloop n'est actuellement pas disponible sous Windows, installez-le avec pip install uvloop .

import asyncio
import uvloop

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

On peut également modifier la fabrique de boucles d'événements en définissant EventLoopPolicy sur celle d' uvloop .

import asyncio
import uvloop

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

Primitive de synchronisation: événement

Concept

Utilisez un Event pour synchroniser la planification de plusieurs coroutines .

En d'autres termes, un événement est comme un coup de feu lors d'une course à pied: il permet aux coureurs de quitter les starting-blocks.

Exemple

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)

Sortie:

Consommateur B en attente
Un consommateur en attente
SET D'ÉVÉNEMENTS
Consommateur B déclenché
Consommateur A déclenché

Un websocket simple

Ici, nous faisons un websocket echo simple en utilisant asyncio . Nous définissons des coroutines pour se connecter à un serveur et envoyer / recevoir des messages. Les communications du websocket sont exécutées dans une coroutine main , qui est exécutée par une boucle d'événement. Cet exemple est modifié depuis un article précédent .

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

Idée commune à propos de l'asyncio

L'idée fausse la plus commune à propos de asnycio est asnycio qu'il vous permet d'exécuter n'importe quelle tâche en parallèle, en contournant le GIL (Global Interpreter Lock) et donc en exécutant des jobs de blocage en parallèle (sur des threads séparés). ça ne marche pas !

asyncio (et les bibliothèques asyncio pour collaborer avec asyncio ) asyncio sur des coroutines: des fonctions qui (en collaboration) renvoient le flux de contrôle à la fonction appelante. Notez asyncio.sleep dans les exemples ci-dessus. Voici un exemple de coroutine non bloquante qui attend en arrière-plan et redonne le contrôle à la fonction appelante (lorsqu'elle est appelée avec await ). time.sleep est un exemple de fonction de blocage. le flux d'exécution du programme s'arrêtera juste là et ne reviendra qu'après que time.sleep soit terminé.

un exemple réel est la bibliothèque de requests qui consiste (pour le moment) en fonctions de blocage uniquement. il n'y a pas de concurrence si vous appelez l'une de ses fonctions dans asyncio . aiohttp d'autre part a été construit avec asyncio à l'esprit. ses coroutines seront concurrentes.

• Si vous avez des tâches liées au processeur qui asyncio longtemps, que vous souhaitez exécuter en parallèle asyncio n'est pas pour vous. pour cela, vous avez besoin de threads ou de multiprocessing .

• Si vous avez des travaux liés aux E / S en cours d'exécution, vous pouvez les exécuter simultanément en utilisant asyncio .