为何asyncio.sleep表现得像阻塞函数？异步代码执行异常排查

你现在的代码是串行执行的——在for循环里每次await store(...)都会等待当前协程完全执行完，才会进入下一次循环。哪怕store里用了asyncio.sleep，await会让整个run_tests暂停，直到这个store跑完，所以看起来和time.sleep没区别，完全没用到异步并发的特性。

你的需求是让多个store协程同时运行，同时在它们完成时更新last_finished（只统计无延迟的任务），最后断言last_finished是"are"。

步骤1：补全缺失的变量定义 #

原代码里没定义DB_FILE，先补上：

DB_FILE = "test_db.txt"

步骤2：用异步任务实现并发执行 #

不要在循环里直接await，而是用asyncio.create_task创建异步任务，让它们在后台运行；同时绑定每个任务对应的text和delay标记，任务完成时更新last_finished。

修改后的完整代码：

import asyncio
import os
import time

DB_FILE = "test_db.txt"

async def store(text, delay: bool = False):
    mode = "a" if os.path.exists(DB_FILE) else "w"
    if delay:
        await asyncio.sleep(5)
    with open(DB_FILE, mode=mode) as f:
        timestamp = str(time.time())
        row = f"{text}\t{timestamp}\n"
        f.write(row)
        f.flush()
        print(f"done saving: {text}")

async def run_tests():
    initial_text = "hi"
    await store(initial_text)
    last_finished = initial_text
    
    tasks = []
    # 遍历任务列表，创建异步任务并保存对应信息
    for text, delay in [("how", False), ("are", False), ("you", True)]:
        task = asyncio.create_task(store(text, delay))
        tasks.append((task, text, delay))
    
    # 逐个等待任务完成，更新last_finished
    for task, text, delay in tasks:
        await task
        if not delay:
            last_finished = text
    
    assert last_finished == "are"
    print("断言通过！")

if __name__ == "__main__":
    # 可选：每次测试前清空文件，避免历史数据干扰
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
    asyncio.run(run_tests())

关键说明 #

• asyncio.create_task会把store协程加入事件循环后台运行，不会阻塞当前run_tests的执行（直到主动用await等待它）。
• 这里先创建所有任务再逐个等待，三个store协程会并发执行：无延迟的"how"和"are"会快速完成，带延迟的"you"会在5秒后完成，最终last_finished会被更新为"are"，符合预期。

更灵活的实时监听写法 #

如果想在任务完成时立刻更新last_finished，不用等所有任务创建完再等待，可以用回调逻辑：

async def run_tests():
    initial_text = "hi"
    await store(initial_text)
    last_finished = initial_text
    
    async def on_task_done(task, text, delay):
        await task
        nonlocal last_finished
        if not delay:
            last_finished = text
    
    tasks = []
    for text, delay in [("how", False), ("are", False), ("you", True)]:
        task = asyncio.create_task(store(text, delay))
        # 创建监听任务，任务完成时触发更新
        tasks.append(asyncio.create_task(on_task_done(task, text, delay)))
    
    # 等待所有监听任务完成
    await asyncio.gather(*tasks)
    
    assert last_finished == "are"
    print("断言通过！")

这样三个store任务一创建就开始运行，只要无延迟的任务完成，last_finished就会立刻更新，无需按顺序等待。

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