含阻塞与异步混合代码的函数无法用run_in_executor运行的解决方法

问题原因 #

你提供的错误代码中，blocking_function是异步函数（带async关键字），但直接把它传给run_in_executor——这个方法只能接受同步函数，调用异步函数只会得到一个未执行的协程对象，既不会触发函数逻辑，还会抛出“coroutine was never awaited”的警告。同时，异步函数里的await必须在事件循环中执行，不能直接在线程池里运行整个异步函数。

解决方案 #

核心思路是：把函数中的阻塞代码片段单独剥离，委托给线程池执行，异步代码留在事件循环中调度，同时用asyncio.gather实现多任务并发。

方案1：拆分阻塞与异步逻辑（兼容所有Python版本）

将阻塞部分抽成独立的同步函数，用run_in_executor执行并await结果：

import asyncio
import time

# 单独提取阻塞逻辑为同步函数
def sync_blocking_task():
    print("Blocking function called")
    time.sleep(1)
    print("Blocking function finished")

async def mixed_function():
    for x in range(4):
        # 在线程池中执行阻塞任务，await获取结果
        await asyncio.get_event_loop().run_in_executor(None, sync_blocking_task)
        print("Async code running:")
        await asyncio.sleep(1)
        print("Async code finished")

async def concurrent_function():
    for x in range(10):
        print(x)
        await asyncio.sleep(1)

async def main():
    print("Main function called")
    # 并发运行两个异步任务
    await asyncio.gather(mixed_function(), concurrent_function())
    print("Main function finished")

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

方案2：用asyncio.to_thread简化代码（Python 3.9+）

Python 3.9新增的asyncio.to_thread可以更简洁地将同步代码放到线程中执行，无需单独拆分函数：

import asyncio
import time

async def mixed_function():
    for x in range(4):
        print("Blocking function called")
        # 直接用to_thread执行阻塞的sleep
        await asyncio.to_thread(time.sleep, 1)
        print("Blocking function finished")
        print("Async code running:")
        await asyncio.sleep(1)
        print("Async code finished")

async def concurrent_function():
    for x in range(10):
        print(x)
        await asyncio.sleep(1)

async def main():
    print("Main function called")
    await asyncio.gather(mixed_function(), concurrent_function())
    print("Main function finished")

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

关键说明 #

• 两种方案都是让阻塞代码在单独线程中运行，避免阻塞主事件循环，保证concurrent_function能同时执行。
• 用asyncio.gather替代原来的“先丢执行器再await单个任务”的写法，能更清晰地实现多任务并发调度。

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