在asyncio程序中使用time.sleep()的问题及替代方案咨询

你的核心问题是asyncio事件循环被阻塞操作（如time.sleep）卡住——time.sleep会占用事件循环所在线程，完全阻止其他异步任务执行。之前用create_task和gather没用，是因为这些方法仅负责调度异步任务，但任务内部的阻塞代码仍会垄断线程资源。

下面分两种场景给出解决方案：

场景1：可以修改含time.sleep的函数 #

如果能控制阻塞代码（比如你自己写的test_loop），直接把time.sleep()替换成await asyncio.sleep()，这是最贴合asyncio设计的方案：

import asyncio

async def test_loop():
    loop_counter = 0
    while True:
        print('loop')
        # 替换time.sleep为asyncio.sleep，主动让出事件循环控制权
        await asyncio.sleep(2)
        if loop_counter >= 5:
            break
        loop_counter += 1

async def delay_msgs():
    print('run msgs')
    await asyncio.sleep(5)
    print('asdf')
    await asyncio.sleep(10)
    print("qwerty")

async def main():
    task1 = asyncio.create_task(delay_msgs())
    task2 = asyncio.create_task(test_loop())
    await asyncio.gather(task1, task2)

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

修改后两个任务会真正并发执行：test_loop每2秒打印一次，delay_msgs会在5秒后正常输出asdf，不会被test_loop的sleep操作卡住。

场景2：无法修改阻塞函数（如第三方库的openai.ChatCompletion.create） #

如果阻塞代码是第三方库（比如openai的方法内部使用了time.sleep），无法直接修改，就把这个阻塞操作放到线程池里执行，让事件循环线程不被占用。

方案A：用asyncio.to_thread()（Python 3.9+） #

这是最简单的实现方式，直接把阻塞函数包装在to_thread中：

import asyncio
import time

# 模拟无法修改的阻塞函数（对应openai.ChatCompletion.create）
def blocking_test_loop():
    loop_counter = 0
    while True:
        print('loop')
        time.sleep(2)
        if loop_counter >= 5:
            break
        loop_counter += 1

async def test_loop():
    # 将阻塞函数放到线程池执行
    await asyncio.to_thread(blocking_test_loop)

async def delay_msgs():
    print('run msgs')
    await asyncio.sleep(5)
    print('asdf')
    await asyncio.sleep(10)
    print("qwerty")

async def main():
    task1 = asyncio.create_task(delay_msgs())
    task2 = asyncio.create_task(test_loop())
    await asyncio.gather(task1, task2)

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

方案B：用loop.run_in_executor()（兼容Python 3.7+） #

如果你的Python版本低于3.9，可手动指定线程池执行阻塞函数：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_test_loop():
    loop_counter = 0
    while True:
        print('loop')
        time.sleep(2)
        if loop_counter >= 5:
            break
        loop_counter += 1

async def test_loop():
    loop = asyncio.get_running_loop()
    # 调用线程池执行阻塞函数
    await loop.run_in_executor(ThreadPoolExecutor(), blocking_test_loop)

async def delay_msgs():
    print('run msgs')
    await asyncio.sleep(5)
    print('asdf')
    await asyncio.sleep(10)
    print("qwerty")

async def main():
    task1 = asyncio.create_task(delay_msgs())
    task2 = asyncio.create_task(test_loop())
    await asyncio.gather(task1, task2)

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

原理说明 #

asyncio的事件循环基于单线程调度，所有异步任务都在该线程内切换执行。如果某个任务执行阻塞操作（如time.sleep、同步IO），整个事件循环会被卡住，直到阻塞结束。将阻塞操作放到线程池后，事件循环线程可继续处理其他任务，等线程池内的阻塞操作完成后再获取结果，从而保证异步特性不受影响。

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