如何使用 asyncio 并行化阻塞代码答案

【问题标题】：How to parallelise blocking code with asyncio如何使用 asyncio 并行化阻塞代码
【发布时间】：2020-06-09 03:33:19
【问题描述】：

我知道 StackOverflow 中的 asyncio 功能很多，但尽管这里回答了很多问题，但我仍然不明白如何做一些简单的事情，比如并行化 2 个执行阻塞代码的任务。

例如，这很好用：

import asyncio


async def slow_thing():
    await asyncio.sleep(2)


async def try_alpha():
    print("Alpha start")
    await slow_thing()
    print("Alpha stop")
    return "Alpha"


async def try_bravo():
    print("Bravo start")
    await slow_thing()
    print("Bravo stop")
    return "Bravo"


async def main():
    futures = [
        try_alpha(),
        try_bravo(),
    ]
    for response in await asyncio.gather(*futures):
        print(response)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

输出正是我想要的：

Alpha start
Bravo start
*2 second wait*
Alpha stop
Bravo stop
Alpha
Bravo

但是，如果我将await syncio.sleep(2) 换成time.sleep(2)，输出就好像我的代码没有任何异步：

Alpha start
*2 second wait*
Alpha stop
Bravo start
*2 second wait*
Bravo stop
Alpha
Bravo

问题是，在我的真实示例中，我无法控制那些慢代码，因此我无法将其更改为使用协程。在某些情况下，这只是 requests.get() 的一堆用途，而在其他情况下，我正在使用 kodijson 库，它做了很多我无法访问的事情。

所以我想知道 asyncio 是否是正确的工具。当您尝试与 .gather() 并行时，是否可以在异步代码中使用阻塞代码？

另外请注意，我（不幸地）在这方面坚持使用 Python 3.6。我正在写一个Mycroft 扩展，这就是他们目前所坚持的环境。

【问题讨论】：

对于阻塞呼叫，使用loop.run_in_executor
另外请注意，如果您没有一堆代码可以在 asyncio 之上利用异步 I/O，那么它可能不是您的应用程序的正确选择，并且你应该直接使用threading/multiprocessing/concurrent.futures。
我用loop.run_in_executor(None, time.sleep, 2) 替换了sleep 行，它有点有效。它在应该等待的地方等待，但也会打印出错误：RuntimeError: Event loop is closed
但是您使用concurrent.futures 的建议看起来可能对我有用，谢谢！
This answer 展示了如何将concurrent.futures 直接用于此类任务。

标签： python python-asyncio

【解决方案1】：

只有当有东西等待时，协程才能“并行”做一些事情。例如，在您上面的代码中，它与 asyncio.sleep 一起工作的原因是您可以在其上等待它。您只能等待为此目的而设计的特定功能。这就是标准 time.sleep 不起作用的原因，因为您不能使用关键字 await 。 requests 库也是如此。

幸运的是，您可以使用美妙的 aiohttp 库：https://docs.aiohttp.org 这将为您提供同时发出多个请求所需的确切信息。

【讨论】：

这个答案是正确的，但是 OP 明确表示他们不控制“慢”代码，即它不能仅仅从 requests 移植到 aiohttp。如 cmets 所述，这可以使用 run_in_executor 修复，但也可以直接使用 concurrent.futures。

【解决方案2】：

在我在这里以 cmets 的形式获得帮助后，我能够使用 concurrent.futures 汇总一个解决方案：

import concurrent.futures
import time


def slow_1(s):
    time.sleep(5)
    print(f"1: {s}")
    return "1: ok"


def slow_2(s):
    time.sleep(1)
    print(f"2: {s}")
    return "2: ok"


def slow_3(s):
    time.sleep(1)
    print(f"3: {s}")
    return "3: ok"


with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = (
        executor.submit(slow_1, "x"),
        executor.submit(slow_2, "y"),
        executor.submit(slow_3, "z"),
    )
    concurrent.futures.wait(futures)
    for future in futures:
        try:
            print(future.result())
        except:  # This should obviously be more explicit
            pass

哪些输出：

2: y
3: z
1: x
1: ok
2: ok
3: ok

我应该注意到，官方文档并不清楚你可以通过在未来调用 .result() 从函数中获取返回值，或者你需要遍历 futures 值来得到所说的结果。 .wait() 返回一个 done 和 not_done 值的元组按照它们返回的顺序，所以循环 done 的值对我来说破坏了很多东西。如果你像我一样，只想一次做 3 件缓慢的事情并从这三件事情中获得结果，那么这段代码可能适合你。

【讨论】：