如何将asyncio与现有的阻止库一起使用？

Question

我的阻塞功能很少foo，bar我无法更改（某些我无法控制的内部库。与一个或多个网络服务进行对话）。如何将其用作异步？例如，我将不执行以下操作。

results = []
for inp in inps:
    val = foo(inp)
    result = bar(val)
    results.append(result)

这将是低效的，因为我foo在等待第一个输入的同时可以调用第二个输入bar。我如何包装它们以便它们可与asyncio一起使用（即new
async，await语法）？

让我们假设函数是可重入的。也就是说，foo如果已经foo处理过先前的事件，则可以再次调用。

更新资料

使用可重用的装饰器扩展答案。单击此处为例。

def run_in_executor(f):
    @functools.wraps(f)
    def inner(*args, **kwargs):
        loop = asyncio.get_running_loop()
        return loop.run_in_executor(None, functools.partial(f, *args, **kwargs))

    return inner

Answer 1

这里有两个问题：首先，如何异步运行阻塞代码，其次，如何并行运行异步代码（异步是单线程的，因此GIL仍然适用，因此不是 真正的 并发，但我离题了。

可以使用asyncio.ensure_future创建并行任务，如此处所述。

要运行同步代码，您将需要在executor中运行阻塞代码。例：

import concurrent.futures
import asyncio
import time

def blocking(delay):
    time.sleep(delay)
    print('Completed.')

async def non_blocking(loop, executor):
    # Run three of the blocking tasks concurrently. asyncio.wait will
    # automatically wrap these in Tasks. If you want explicit access
    # to the tasks themselves, use asyncio.ensure_future, or add a
    # "done, pending = asyncio.wait..." assignment
    await asyncio.wait(
        fs={
            # Returns after delay=12 seconds
            loop.run_in_executor(executor, blocking, 12),

            # Returns after delay=14 seconds
            loop.run_in_executor(executor, blocking, 14),

            # Returns after delay=16 seconds
            loop.run_in_executor(executor, blocking, 16)
        },
        return_when=asyncio.ALL_COMPLETED
    )

loop = asyncio.get_event_loop()
executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
loop.run_until_complete(non_blocking(loop, executor))

如果要使用for循环调度这些任务（如您的示例中所示），则有几种不同的策略，但是基本方法是使用for循环（或列表理解等） 调度
任务，并通过asyncio等待它们。等待， 然后 检索结果。例：

done, pending = await asyncio.wait(
    fs=[loop.run_in_executor(executor, blocking_foo, *args) for args in inps],
    return_when=asyncio.ALL_COMPLETED
)

# Note that any errors raise during the above will be raised here; to
# handle errors you will need to call task.exception() and check if it
# is not None before calling task.result()
results = [task.result() for task in done]