为何anyio最多同时运行40个任务，100个任务预期5秒完成却耗时15秒？

问题原因 #

• 核心限制来自AnyIO to_thread.run_sync 默认的全局线程池大小上限：AnyIO默认将同步函数转异步运行的线程池最大工作线程数设为 CPU核心数 * 5，若使用8核CPU，默认上限刚好为40，和你观测到的最大并行任务数完全吻合。
• 100个任务按40个/批分批执行，每批耗时5秒，3批总耗时刚好为15秒，和你的运行结果一致。该限制是AnyIO层的统一规则，和你使用的asyncio/trio异步后端无关，因此切换后端不会改变现象。

修复方案 #

你可以通过自定义线程池限流器，扩大允许的最大并行线程数解决该问题，修改步骤如下：

1. 导入CapacityLimiter并创建自定义限流实例，允许最大100个同步线程同时运行：

from anyio import CapacityLimiter

# 配置并行线程上限为100
thread_limiter = CapacityLimiter(total_tokens=100)

2. 修改async_sleep_5_and_print中to_thread.run_sync的调用，传入自定义限流器：

async def async_sleep_5_and_print(receive_other: rstream):
    async with receive_other:
        async with anyio.create_task_group() as task_group:
            async for other in receive_other:
                task_group.start_soon(
                    anyio.to_thread.run_sync, sync_sleep_5_and_print, other,
                    limiter=thread_limiter
                )

可选优化 #

当前代码中创建内存流使用默认无缓冲配置，发送方需要等待接收方取走数据才能继续执行，你可以在创建流时指定缓冲区大小，进一步提升流水线效率：

# 创建缓冲区大小为100的内存流，可一次性容纳所有待处理数据
send, receive = anyio.create_memory_object_stream(max_buffer_size=100)
send_other, receive_other = anyio.create_memory_object_stream(max_buffer_size=100)

修改后100个IO等待型任务可以同时并行运行，总耗时会降到5秒左右。

内容的提问来源于stack exchange，提问作者phi friday