asyncio异步任务调度问题：如何避免40秒阻塞并实现任务依赖

原脚本问题分析 #

1. 数据传递不符合需求：原代码中第三个函数处理的是result1，但需求要求处理result2
2. 任务设计不合理：原third_function采用单循环串行处理队列的方式，虽asyncio.sleep不会阻塞事件循环，但不符合“每次处理完result1后延迟40秒启动处理对应result2”的一对一触发需求

正确实现代码 #

import asyncio
import logging

# 初始化日志配置，方便查看执行时序
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")

async def first_function(queue1, queue2):
    while True:
        # 模拟生成结果
        result1 = "some result"
        result2 = {"key": "value"}
        await queue1.put(result1)
        await queue2.put(result2)
        logging.info("First function generated new results")
        await asyncio.sleep(1)  # 每秒生成一次结果

async def handle_result2(result2):
    # 延迟40秒后处理result2
    await asyncio.sleep(40)
    logging.info(f"Handled result2 after 40s delay: {result2}")

async def second_function(queue1, queue2):
    while True:
        # 等待first_function生成的结果
        result1 = await queue1.get()
        result2 = await queue2.get()
        
        # 处理result1
        logging.info(f"Second function processed result1: {result1}")
        
        # 触发延迟40秒处理result2的任务，不阻塞当前流程
        asyncio.create_task(handle_result2(result2))
        
        await asyncio.sleep(1)  # 模拟处理result1的耗时

async def main():
    queue1 = asyncio.Queue()
    queue2 = asyncio.Queue()
    
    # 创建并发任务
    task1 = asyncio.create_task(first_function(queue1, queue2))
    task2 = asyncio.create_task(second_function(queue1, queue2))
    
    await asyncio.gather(task1, task2)

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

关键修改说明 #

• 新增handle_result2函数专门处理延迟40秒的逻辑，每次由second_function触发独立的异步任务，确保不会阻塞其他流程
• 移除多余的queue3，直接将result2传递给延迟任务，符合需求中“处理result2”的要求
• first_function的循环完全不受延迟任务影响，因为asyncio.sleep会主动让出事件循环控制权，事件循环会继续调度first_function和second_function的任务

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