4. asyncio 库实战：Task 与 Future 的关系、协程调度器、超时控制与取消任务

说实话，asyncio 这块内容，我当年刚接触的时候也绕了不少弯路。尤其是 Task 和 Future 这两个概念，看着像双胞胎，实际用起来差别还挺大。今天咱们就把它彻底掰扯清楚。

4.1 Task 与 Future：到底谁是谁？

先问个问题：Future 是什么？说白了，它是一个「未来结果的占位符」。你启动一个异步操作，立刻拿到一个 Future 对象，但真正的结果还没到。等操作完成了，你把结果塞进去，别人就能从 Future 里取出来。

那 Task 呢？Task 是 Future 的子类，但它更「高级」一点。Task 不光能装结果，它还能管理一个协程的运行。你想想看，协程本身只是一个函数，谁去调度它？谁去驱动它执行？就是 Task。

我在项目中遇到过这样一个场景：有人直接用 loop.create_future() 然后手动 set_result()，结果忘了驱动协程，整个程序卡死了。嗯，这就是没搞清楚 Task 和 Future 的区别。

4.2 协程调度器：事件循环的秘密

asyncio 的调度器，其实就是事件循环（Event Loop）。它怎么工作的？我习惯用一个比喻：事件循环就像一个永不休息的接线员，手里拿着一个任务队列，谁准备好了就喊谁。

import asyncio

async def worker(name, delay):
    print(f"{name} 开始工作")
    await asyncio.sleep(delay)
    print(f"{name} 工作完成")
    return f"{name} 的结果"

async def main():
    # 创建三个任务
    task1 = asyncio.create_task(worker("A", 2))
    task2 = asyncio.create_task(worker("B", 1))
    task3 = asyncio.create_task(worker("C", 3))
    
    # 等待所有任务完成
    results = await asyncio.gather(task1, task2, task3)
    print(results)

asyncio.run(main())

你看，三个任务同时启动，但 B 先完成（因为它只睡 1 秒），然后 A，最后 C。事件循环会不断检查哪个协程的 await 条件满足了，就把它唤醒继续执行。

4.3 超时控制：别让程序死等

我曾经在生产环境踩过一个坑：一个外部 API 调用没有设置超时，结果对方服务挂了，我的整个服务也跟着卡死。从那以后，我养成了一个习惯——所有异步操作必须加超时。

asyncio 提供了两种超时方式：

方式	适用场景	示例
asyncio.wait_for()	单个协程超时	await asyncio.wait_for(coro, timeout=5)
asyncio.timeout() (3.11+)	上下文管理器，更灵活	async with asyncio.timeout(5): ...

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "终于完成了"

async def main():
    try:
        # 只等 3 秒，超时就抛异常
        result = await asyncio.wait_for(slow_operation(), timeout=3)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时了！")

asyncio.run(main())

4.4 取消任务：优雅地停止

取消任务，说白了就是给协程发一个「别干了」的信号。asyncio 用 task.cancel() 来实现，协程内部会收到一个 CancelledError 异常。

import asyncio

async def cancellable_worker():
    try:
        for i in range(10):
            print(f"工作中... {i}")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("我被取消了，正在清理...")
        # 做一些清理工作
        raise  # 重新抛出，表示确认取消

async def main():
    task = asyncio.create_task(cancellable_worker())
    await asyncio.sleep(3)
    task.cancel()  # 发送取消信号
    try:
        await task
    except asyncio.CancelledError:
        print("任务已取消")

asyncio.run(main())

这里有个细节：CancelledError 必须重新抛出，否则 asyncio 会认为任务「拒绝取消」，继续运行。我见过有人忘了 raise，结果任务怎么都停不下来，排查了半天。

4.5 综合实战：一个带超时和取消的任务管理器

好了，咱们把上面这些知识点串起来，写一个实用的任务管理器。这个管理器可以同时运行多个任务，支持超时控制，还能手动取消。

import asyncio
from typing import Dict, Optional

class TaskManager:
    def __init__(self):
        self._tasks: Dict[str, asyncio.Task] = {}
    
    async def run_task(self, name: str, coro, timeout: Optional[float] = None):
        """运行一个任务，支持超时"""
        if name in self._tasks:
            raise ValueError(f"任务 {name} 已存在")
        
        task = asyncio.create_task(coro)
        self._tasks[name] = task
        
        try:
            if timeout:
                result = await asyncio.wait_for(task, timeout=timeout)
            else:
                result = await task
            return result
        except asyncio.TimeoutError:
            print(f"任务 {name} 超时，正在取消...")
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass
            raise
        finally:
            self._tasks.pop(name, None)
    
    def cancel_task(self, name: str):
        """取消指定任务"""
        task = self._tasks.get(name)
        if task and not task.done():
            task.cancel()
    
    async def cancel_all(self):
        """取消所有任务"""
        for name, task in self._tasks.items():
            if not task.done():
                task.cancel()
        # 等待所有任务完成取消
        await asyncio.gather(*self._tasks.values(), return_exceptions=True)
        self._tasks.clear()

# 使用示例
async def example():
    manager = TaskManager()
    
    async def long_task():
        try:
            await asyncio.sleep(10)
            return "完成"
        except asyncio.CancelledError:
            print("long_task 被取消")
            raise
    
    # 运行一个 5 秒超时的任务
    try:
        result = await manager.run_task("task1", long_task(), timeout=5)
        print(result)
    except asyncio.TimeoutError:
        print("任务超时了")
    
    # 手动取消
    task2 = asyncio.create_task(long_task())
    manager._tasks["task2"] = task2
    await asyncio.sleep(1)
    manager.cancel_task("task2")
    await asyncio.sleep(0.5)

asyncio.run(example())

4.6 知识体系总览

下面这张图，是我梳理的 asyncio 任务调度核心逻辑。你看一眼，基本就能把今天讲的内容串起来。

从这张图你能看到：事件循环驱动任务队列，每个 Task 包装一个协程，最终产出 Future 结果。超时控制和取消任务，都是在 Task 层面操作的。调度器（如 gather）则负责协调多个 Task 的执行顺序。

嗯，今天的内容就到这。asyncio 的 Task 和 Future 其实没那么玄乎，你只要记住：Task 是干活的人，Future 是装结果的箱子。超时和取消，就是给干活的人设个闹钟和紧急停止按钮。多写几个例子，自然就熟了。

想要更多资料，联系可可