Можно ли использовать функцию async внутри процесса? [Дубликат]

Вы должны быть в состоянии безопасно объединить asyncio и multiprocessing без особых проблем, хотя вы не должны использовать multiprocessing напрямую. Кардинальный грех asyncio (и любой другой асинхронной структуры на основе цикла событий) блокирует цикл события. Если вы попытаетесь напрямую использовать multiprocessing, каждый раз, когда вы блокируете дождевой процесс, вы собираетесь заблокировать цикл событий. Очевидно, что это плохо.

Самый простой способ избежать этого - использовать BaseEventLoop.run_in_executor для выполнения функции в concurrent.futures.ProcessPoolExecutor . ProcessPoolExecutor - пул процессов, реализованный с использованием multiprocessing.Process, но asyncio имеет встроенную поддержку для выполнения в нем функции без блокировки цикла события. Вот простой пример:

import time
import asyncio
from concurrent.futures import ProcessPoolExecutor

def blocking_func(x):
   time.sleep(x) # Pretend this is expensive calculations
   return x * 5

@asyncio.coroutine
def main():
    #pool = multiprocessing.Pool()
    #out = pool.apply(blocking_func, args=(10,)) # This blocks the event loop.
    executor = ProcessPoolExecutor()
    out = yield from loop.run_in_executor(executor, blocking_func, 10)  # This does not
    print(out)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Для большинства случаев эта функция сама по себе достаточно хороша. Если вам понадобятся другие конструкции из multiprocessing, например Queue, Event, Manager и т. Д., Существует сторонняя библиотека, называемая aioprocessing (полное раскрытие: Я написал это), который предоставляет asyncio -совместимые версии всех структур данных multiprocessing. Вот пример демонстрации того, что:

import time
import asyncio
import aioprocessing
import multiprocessing

def func(queue, event, lock, items):
    with lock:
        event.set()
        for item in items:
            time.sleep(3)
            queue.put(item+5)
    queue.close()

@asyncio.coroutine
def example(queue, event, lock):
    l = [1,2,3,4,5]
    p = aioprocessing.AioProcess(target=func, args=(queue, event, lock, l)) 
    p.start()
    while True:
        result = yield from queue.coro_get()
        if result is None:
            break
        print("Got result {}".format(result))
    yield from p.coro_join()

@asyncio.coroutine
def example2(queue, event, lock):
    yield from event.coro_wait()
    with (yield from lock):
        yield from queue.coro_put(78)
        yield from queue.coro_put(None) # Shut down the worker

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    queue = aioprocessing.AioQueue()
    lock = aioprocessing.AioLock()
    event = aioprocessing.AioEvent()
    tasks = [ 
        asyncio.async(example(queue, event, lock)),
        asyncio.async(example2(queue, event, lock)),
    ]   
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()