Блокирующие код и asyncio

Мы знаем, что мы можем сделать много асинхронных операций с помощью asyncio, но, как выполнить блокирующие операции с ним? На самом деле это довольно просто, asyncio позволяет нам запускать код с блокирующими командами с помощью метода BaseEventLoop.run_in_executor. Он будет выполнять наши функции параллельно и предоставит нам объекты Future, через которые мы можем получить результаты и состояние выполнения (await или yield from).

Давайте посмотрим пример с популярной библиотекой requests:

import asyncio
import requests

loop = asyncio.get_event_loop()


def get_html(url):
    return loop.run_in_executor(None, requests.get, url)


@asyncio.coroutine
def main():
    resp1 = yield from get_html("http://masnun.com")
    resp2 = yield from get_html("http://python.org")

    print(resp2, resp1)


loop.run_until_complete(main())

import asyncio

import requests

loop = asyncio.get_event_loop()

def get_html(url):

return loop.run_in_executor(None, requests.get, url)

@asyncio.coroutine

def main():

resp1 = yield from get_html("http://masnun.com")

resp2 = yield from get_html("http://python.org")

print(resp2, resp1)

loop.run_until_complete(main())

В добавление маленький пример из документации:

import asyncio
import concurrent.futures

def blocking_io():
    # File operations (such as logging) can block the
    # event loop: run them in a thread pool.
    with open('/dev/urandom', 'rb') as f:
        return f.read(100)

def cpu_bound():
    # CPU-bound operations will block the event loop:
    # in general it is preferable to run them in a
    # process pool.
    return sum(i * i for i in range(10 ** 7))

async def main():
    loop = asyncio.get_running_loop()

    ## Options:

    # 1. Run in the default loop's executor:
    result = await loop.run_in_executor(
        None, blocking_io)
    print('default thread pool', result)

    # 2. Run in a custom thread pool:
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, blocking_io)
        print('custom thread pool', result)

    # 3. Run in a custom process pool:
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, cpu_bound)
        print('custom process pool', result)

asyncio.run(main())

import asyncio

import concurrent.futures

def blocking_io():

# File operations (such as logging) can block the

# event loop: run them in a thread pool.

with open('/dev/urandom', 'rb') as f:

return f.read(100)

def cpu_bound():

# CPU-bound operations will block the event loop:

# in general it is preferable to run them in a

# process pool.

return sum(i * i for i in range(10 ** 7))

async def main():

loop = asyncio.get_running_loop()

## Options:

# 1. Run in the default loop's executor:

result = await loop.run_in_executor(

None, blocking_io)

print('default thread pool', result)

# 2. Run in a custom thread pool:

with concurrent.futures.ThreadPoolExecutor() as pool:

result = await loop.run_in_executor(

pool, blocking_io)

print('custom thread pool', result)

# 3. Run in a custom process pool:

with concurrent.futures.ProcessPoolExecutor() as pool:

result = await loop.run_in_executor(

pool, cpu_bound)

print('custom process pool', result)

asyncio.run(main())

Этот код возвращает asyncio.Future объект.

Начиная с версии 3.5.3: loop.run_in_executor () больше не настраивает max_workers исполнителей пула потоков, который он создает, а оставляет пул потоков (ThreadPoolExecutor) в настройках по умолчанию.

Архив

Последнее