Python asyncio并发任务的超时控制与优雅关闭

import asyncio
import time

# 模拟长时间运行或阻塞的网络I/O任务
async def watch_task_data():
    print(f"[{time.time():.2f}] watch_task_data: 启动，模拟等待数据...")
    try:
        # 模拟长时间等待网络数据，可能永不返回
        await asyncio.sleep(100) # 模拟阻塞100秒
        print(f"[{time.time():.2f}] watch_task_data: 收到数据并完成。")
    except asyncio.CancelledError:
        print(f"[{time.time():.2f}] watch_task_data: 任务被取消。")
    except Exception as e:
        print(f"[{time.time():.2f}] watch_task_data: 发生异常: {e}")
    finally:
        print(f"[{time.time():.2f}] watch_task_data: 结束。")

async def watch_task_news():
    print(f"[{time.time():.2f}] watch_task_news: 启动，模拟等待新闻...")
    try:
        # 模拟另一个长时间等待新闻的任务
        await asyncio.sleep(100) # 模拟阻塞100秒
        print(f"[{time.time():.2f}] watch_task_news: 收到新闻并完成。")
    except asyncio.CancelledError:
        print(f"[{time.time():.2f}] watch_task_news: 任务被取消。")
    except Exception as e:
        print(f"[{time.time():.2f}] watch_task_news: 发生异常: {e}")
    finally:
        print(f"[{time.time():.2f}] watch_task_news: 结束。")

async def main():
    tasks = [
        watch_task_data(),
        watch_task_news(),
    ]

    print(f"[{time.time():.2f}] 主程序：开始等待任务，最长等待5秒...")
    # 设置整体超时为5秒
    done, pending = await asyncio.wait(tasks, timeout=5)

    print(f"[{time.time():.2f}] 主程序：等待结束。已完成任务数: {len(done)}, 未完成任务数: {len(pending)}")

    # 1. 处理已完成的任务 (done 集合)
    # 这些任务可能已正常完成，也可能在超时前抛出异常
    for task in done:
        try:
            result = task.result() # 获取任务结果，如果任务抛出异常，这里会重新抛出
            print(f"[{time.time():.2f}] 已完成任务结果: {result if result is not None else '无'}")
        except asyncio.CancelledError:
            # 理论上，done集合中的任务不应该被取消，除非在wait返回前被外部取消
            print(f"[{time.time():.2f}] 已完成任务被取消 (异常情况)")
        except Exception as e:
            print(f"[{time.time():.2f}] 已完成任务发生异常: {e}")

    # 2. 处理未完成的任务 (pending 集合)
    # 这些任务在超时时仍未完成，通常需要显式取消它们以释放资源
    for task in pending:
        print(f"[{time.time():.2f}] 主程序：正在取消未完成任务: {task.get_name() if hasattr(task, 'get_name') else task}")
        task.cancel() # 发送取消信号

        try:
            # 推荐等待任务真正结束，以确保任务有机会执行清理逻辑
            await task
        except asyncio.CancelledError:
            print(f"[{time.time():.2f}] 未完成任务 {task.get_name() if hasattr(task, 'get_name') else task} 已确认取消。")
        except Exception as e:
            print(f"[{time.time():.2f}] 取消未完成任务 {task.get_name() if hasattr(task, 'get_name') else task} 时发生异常: {e}")

    print(f"[{time.time():.2f}] 主程序：所有任务处理完毕。")

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

async def limited_task():
    try:
        await asyncio.wait_for(some_long_running_coroutine(), timeout=10)
        print("limited_task: 任务在10秒内完成。")
    except asyncio.TimeoutError:
        print("limited_task: 任务超时。")