Python Language
Moduł Asyncio

Składnia coroutine i delegacji

Przed wydaniem Python asyncio moduł asyncio używał generatorów do naśladowania wywołań asynchronicznych, a zatem miał inną składnię niż bieżąca wersja 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)

Asynchroniczni wykonawcy

Uwaga: Używa asynchronicznej / oczekującej składni Python 3.5+

asyncio obsługuje użycie obiektów Executor znalezionych w asyncio concurrent.futures do asynchronicznego planowania zadań. Pętle zdarzeń mają funkcję run_in_executor() która pobiera obiekt Executor , Callable i parametry Callable.

Planowanie zadania dla 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))

Każda pętla zdarzeń ma również „domyślne” gniazdo Executor wykonującego, które można przypisać do Executor . Aby przypisać Executor i zaplanować zadania z pętli, użyj metody 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))

Istnieją dwa główne typy Executor w ProcessPoolExecutor concurrent.futures ThreadPoolExecutor i ProcessPoolExecutor . ThreadPoolExecutor zawiera pulę wątków, które można albo ręcznie ustawić za pomocą konstruktora na określoną liczbę wątków, albo domyślnie na liczbę rdzeni na maszynie razy 5. ThreadPoolExecutor używa puli wątków do wykonywania przypisanych do niej zadań i jest ogólnie lepiej w operacjach związanych z procesorem niż operacjach we / wy. ProcessPoolExecutor z ProcessPoolExecutor który spawnuje nowy proces dla każdego przypisanego mu zadania. ProcessPoolExecutor może przyjmować tylko zadania i parametry, które można wybierać. Najczęstsze zadania, których nie można wybrać, to metody obiektów. Jeśli musisz zaplanować metodę obiektu jako zadanie w Executor , musisz użyć ThreadPoolExecutor .

Korzystanie z UVLoop

uvloop jest implementacją asyncio.AbstractEventLoop opartą na libuv (Używany przez nodejs). Jest zgodny z 99% funkcji asyncio i jest znacznie szybszy niż tradycyjny asyncio.EventLoop . uvloop jest obecnie dostępny w systemie Windows, zainstaluj go za pomocą pip install uvloop .

import asyncio
import uvloop

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

Można również zmienić fabrykę pętli zdarzeń, ustawiając EventLoopPolicy na tę w uvloop .

import asyncio
import uvloop

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

Prymitywna synchronizacja: zdarzenie

Pojęcie

Użyj Event aby zsynchronizować planowanie wielu korporacji .

Mówiąc prościej, wydarzenie jest jak strzał z pistoletu w biegający wyścig: pozwala biegaczom zejść z bloków startowych.

Przykład

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)

Wynik:

Klient B. czeka
Konsument Oczekiwanie
ZESTAW WYDARZEŃ
Uruchomił się konsument B.
Uruchomiony konsument A.

Prosty moduł sieciowy

Tutaj tworzymy prosty echo websocket za pomocą asyncio . Definiujemy coroutines do łączenia się z serwerem i wysyłania / odbierania wiadomości. Połączenia websocket są uruchamiane w main korporacji, która jest uruchamiana przez pętlę zdarzeń. Ten przykład został zmodyfikowany z poprzedniego postu .

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

Częste nieporozumienie dotyczące asyncio

Prawdopodobnie najczęstszym błędem o asnycio jest to, że pozwala na uruchamianie dowolnego zadania równolegle - ominięcia GIL (Global Interpreter Lock), a tym samym utrudnianie wykonania zadania równolegle (na oddzielnych wątków). tak nie jest !

asyncio (i biblioteki zbudowane do współpracy z asyncio ) budują na coroutines: funkcje, które (wspólnie) przywracają kontrolę do funkcji wywołującej. zwróć uwagę na asyncio.sleep w powyższych przykładach. jest to przykład nieblokującej się coroutine, która czeka „w tle” i przekazuje kontrolę z powrotem do funkcji wywołującej (gdy jest wywoływana z await ). time.sleep jest przykładem funkcji blokowania. przepływ wykonania programu po prostu się tam zatrzyma i powróci dopiero po zakończeniu time.sleep .

prawdziwym przykładem jest biblioteka requests która na razie polega wyłącznie na blokowaniu funkcji. nie ma współbieżności, jeśli wywołasz dowolną z jej funkcji w asyncio . aiohttp drugiej strony asyncio został zbudowany z asyncio o asyncio . jego coroutines będą działać jednocześnie.

• jeśli masz długoterminowe zadania związane z procesorem, które chcesz uruchomić równolegle asyncio nie jest dla ciebie. do tego potrzebujesz threads lub multiprocessing .

• jeśli masz IO-bound pracy z systemem, można uruchomić je jednocześnie za pomocą asyncio .