Python异步方法未按预期执行，请求排查修正

你的代码核心问题是误用了同步阻塞的time.sleep()，它会直接卡住异步事件循环，导致所有任务串行等待，无法实现真正的并发。要达成每个Item独立并发、a_task完成后再执行other_task的目标，需要调整异步休眠方式，同时保留每个Item内部的执行顺序。

修改后的代码 #

import asyncio
import pandas as pd


async def execute_check():
    print("execute_check")
    items = [1, 2, 3, 4]
    # 并发启动所有Item的处理任务
    tasks = [fetch_values_and_process(item) for item in items]
    await asyncio.gather(*tasks)


async def fetch_values_and_process(item):
    print(f"fetch_values_and_process for Item {item}")
    # 等待当前Item的fetch完成，不阻塞其他Item的任务
    values_df = await fetch_values(item)

    a_values = values_df[values_df["Label"] == "A"]
    other_values = values_df[values_df["Label"] != "A"]

    if not a_values.empty:
        # 并发执行当前Item下的所有a_task
        a_tasks = [a_task(row, item) for _, row in a_values.iterrows()]
        await asyncio.gather(*a_tasks)
    
    # 必须等当前Item的a_task全部完成后，才执行other_task
    await other_task(other_values, item)


async def fetch_values(item):
    print(f"fetch_values for Item {item}")
    # 用异步休眠替代同步sleep，让出事件循环给其他任务
    await asyncio.sleep(5)
    return pd.DataFrame({"Item": [1, 2, 3, 4], "Label": ["A", "B", "C", "D"]})


async def a_task(row, item):
    await asyncio.sleep(2)
    print(f"a_task finished for Item {item}")


async def other_task(other_values, item):
    await asyncio.sleep(2)
    print(f"other_task finished for Item {item}")


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

关键修改点 #

• 替换time.sleep()为await asyncio.sleep()：time.sleep()会阻塞整个事件循环，让所有异步任务暂停；而asyncio.sleep()是异步休眠，会主动让出CPU，让事件循环处理其他待执行任务，这是实现异步并发的核心。
• 保留每个Item内部的执行顺序：在fetch_values_and_process中，先通过await fetch_values获取当前Item的数据，然后并发执行该Item的所有a_task，等全部完成后再执行other_task，确保同一Item的other_task在a_task之后启动。
• 添加Item标识：在打印信息中加入Item编号，方便观察每个任务对应的Item执行情况。

预期执行行为 #

修改后，四个Item的fetch_values会几乎同时启动（事件循环可并发处理），每个Item的fetch_values完成后会立即启动自己的a_task，等该Item的a_task全部完成后再执行other_task，最终输出类似：

execute_check
fetch_values_and_process for Item 1
fetch_values for Item 1
fetch_values_and_process for Item 2
fetch_values for Item 2
fetch_values_and_process for Item 3
fetch_values for Item 3
fetch_values_and_process for Item 4
fetch_values for Item 4
# 等待约5秒后，四个Item的fetch完成，开始执行a_task
a_task finished for Item 1
a_task finished for Item 2
a_task finished for Item 3
a_task finished for Item 4
# 每个Item的a_task完成后立即启动other_task
other_task finished for Item 1
other_task finished for Item 2
other_task finished for Item 3
other_task finished for Item 4

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