为什么我的函数不按异步方式执行（asyncio to_thread）？

Question

我在看了这个视频后，想让我的代码运行得更快。
我有一个同步的获取函数（这里用mySynchronousFunction表示），我想在一个列表中的多个值上进行评估。
我是在一个jupyter notebook里写这些代码。

下面是一个最小可重现示例：

# Cell one
import asyncio
import pandas as pd
import requests
from time import sleep

# cell two
def mySynchronousFunction():
    sleep(4)
    return(True)

async def turningMyFunctionAsync(string):
    print(f"starting job for {string}")
    returnVal = await asyncio.to_thread(mySynchronousFunction)
    print(f"end of job for {string}")
    return(returnVal)

# Cell tree
myList = [{'val':'val1'},{'val':'val2'},{'val':'val3'},{'val':'val4'},{'val':'val5'}]
async def mainAsyncFunction():
    for val in myList:
        val['result'] = await asyncio.create_task(turningMyFunctionAsync(val['val']))

# Cell four
await mainAsyncFunction()

# Prints:
# starting job for val1
# end of job for val1
# starting job for val2
# end of job for val2
# starting job for val3
# end of job for val3
# starting job for val4
# end of job for val4
# starting job for val5
# end of job for val5

我原本期待第四个单元格会先打印出所有“为val1启动作业”的信息（也就是说，应该同时启动所有的进程），考虑到有很长的等待时间。但实际上，所有的代码都是同步运行的。
我哪里做错了呢？

谢谢。

Answer 1

问题出在 mainAsyncFunction 里的 for 循环。这个 for 循环让你的任务变成了同步的，因为在每次循环中，你都在等待任务完成。也就是说，在每次循环里，你先创建任务，然后等待它完成，这样就意味着你在另一个线程中启动了 mySynchronousFunction，然后获取它的值，所有这些都在继续下一个循环之前完成。

另外，你的代码实际上是不能直接运行的。你不能在协程外面使用 await。我认为唯一允许这样做的地方是 Python 的 REPL（交互式解释器），这可能是为了方便测试而特别处理的。

解决这个问题有几种方法。可以使用 async for，但老实说，我也不知道怎么用。还有 TaskGroups。在这种情况下，我创建了一个简单的解决方案。我们可以简单地建立一个协程的列表。然后，调用 asyncio.gather 来处理这个列表，如果它还不是任务的话，就把每个协程变成任务。需要注意的是，直接运行协程和任务之间的区别在于，任务是并发执行的。asyncio.gather 只是启动协程作为并发任务的一种方式。另外，asyncio.gather 不接受列表作为参数，所以需要用 *coroutines 来把列表作为逗号分隔的参数传入。换句话说，asyncio.gather(coro1, coro2, coro3) 和 asyncio.gather(*coros) 是等价的，其中 coros = [coro1, coro2, coro3]。最后，asyncio.gather 会返回一个任务结果的列表，顺序和传入 asyncio.gather 的顺序一致。

async def mainAsyncFunction():
    coroutines = [turningMyFunctionAsync(val["val"]) for val in myList]
    results = await asyncio.gather(*coroutines)
    print(f"Results: {results}")
    return results

asyncio.run(mainAsyncFunction())

这会输出：

starting job for val1
starting job for val2
starting job for val3
starting job for val4
starting job for val5
end of job for val1
end of job for val2
end of job for val3
end of job for val4
end of job for val5
Results: [True, True, True, True, True]

我最后想说的是，asyncio 并不一定会让你的代码更快。它让代码变得并发。在使用纯协程和任务而不依赖线程的情况下，asyncio 可能会更快，因为它是单线程的，从而避免了线程切换的开销。然而，在你的情况下，你是依赖于线程的。默认情况下，asyncio.to_thread 使用 ThreadPoolExecutor，我认为它默认创建的线程池线程数和核心数相同，所以确实节省了一些创建线程的开销，但它仍然会在这些线程之间切换。