如何在asyncio的不同线程中运行loop_in_executor执行同步函数

代码错误点 #

• 初始asyncio方案错误：
  1. main是同步函数，不能直接使用await关键字，必须定义为异步函数后通过asyncio.run启动
  2. run_sync_in_executor中调用loop.run_in_executor没有加await，拿到的是未执行的协程对象，不是同步方法的实际返回值
  3. 此前对错误原因的理解有误：asyncio所有协程共享同一事件循环是正常设计，报错和gather特性、事件循环共享完全无关，均为代码写法错误导致
• 后续ThreadPoolExecutor方案错误：
  1. run_sync_in_executor是异步函数，不能直接传给executor.submit执行，线程池只能执行同步可调用对象，传入异步函数只会返回未执行的协程对象
  2. ThreadPoolExecutor的并发数参数是max_workers，不是你写的num_workers，参数名错误也会触发报错

最优实现方案 #

方案1：保留异步流程（适合需要在同步方法执行后做异步逻辑处理的场景） #

完全符合初始设计思路的正确写法：

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 原有同步方法
def sync_method(param1, param2):
    # Complex method logic
    return f"Completed: {param1} + {param2}"

async def run_sync_in_executor(param1, param2, pool=None):
    loop = asyncio.get_event_loop()
    # 增加await拿到同步方法的实际返回值
    value = await loop.run_in_executor(pool, sync_method, param1, param2)
    # 此处可以增加对value的后续修改逻辑
    value = value + " (processed)"
    return value

async def main():
    params_list = [[1, 2], [2, 3], [3, 4], [4, 5], [5, 6], [6, 7], [7, 8], [8, 9], [9, 10]]
    # 自定义8线程池，也可以不传pool参数直接使用asyncio内置的默认线程池
    with ThreadPoolExecutor(max_workers=8) as pool:
        output = await asyncio.gather(*[run_sync_in_executor(v[0], v[1], pool) for v in params_list])
    print(output)

if __name__ == "__main__":
    asyncio.run(main())

方案2：纯线程池实现（适合不需要异步逻辑的场景，更简洁） #

如果只是需要并发执行同步方法，完全不需要引入asyncio，直接用线程池即可：

from concurrent.futures import ThreadPoolExecutor

def sync_method(param1, param2):
    # Complex method logic
    return f"Completed: {param1} + {param2}"

def main():
    params_list = [[1, 2], [2, 3], [3, 4], [4, 5], [5, 6], [6, 7], [7, 8], [8, 9], [9, 10]]
    with ThreadPoolExecutor(max_workers=8) as executor:
        # 批量提交所有任务
        results = executor.map(lambda p: sync_method(*p), params_list)
        for res in results:
            # 此处可以增加对返回值的修改逻辑
            print(res)

if __name__ == "__main__":
    main()

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