第 53 课：异步编程：任务（Tasks）与事件循环

学习目标

理解事件循环（Event Loop）在异步编程中的核心作用
掌握如何创建和管理异步任务（Task）
学会使用 asyncio.gather() 和 asyncio.wait() 并发执行多个任务
了解任务的生命周期与常见操作（如取消、等待）
能够区分同步与异步编程的执行流程差异

核心概念

事件循环：异步程序的"总调度师"

想象一个餐厅的后厨：只有一位厨师（CPU），但需要处理多张订单（多个异步任务）。厨师不会在一道菜需要慢炖时傻等，而是会先去处理其他切菜、翻炒的工作，等炖菜时间到了再回来处理。事件循环就是这个厨师，它不断检查哪些任务已准备好执行，然后运行它们。

在 Python 中，asyncio 的事件循环负责：

调度和执行协程
管理 I/O 操作
处理任务间的切换

任务（Task）：事件循环中的"工作单元"

任务是对一个协程的进一步封装，它将协程注册到事件循环中，使其能够被调度执行。当创建一个任务时，相当于告诉事件循环："请帮我运行这个协程，我会在后面需要结果时再处理它。"

关键特性：

任务可以并发运行（在 I/O 等待时切换）
任务会保存协程的执行状态
任务可以被取消或等待完成

任务的状态

任务有三种主要状态：

Pending：等待运行
Running：正在执行
Done：已完成（包括正常完成、被取消或抛出异常）

代码示例

示例1：基础任务创建与运行

import asyncio
import time

async def fetch_data(url, delay):
    """模拟从网络获取数据"""
    print(f"开始获取 {url} 的数据...")
    await asyncio.sleep(delay)  # 模拟I/O操作
    print(f"完成获取 {url} 的数据，耗时 {delay}秒")
    return {"url": url, "data": f"来自 {url} 的数据"}

async def main():
    print("=== 基础任务创建 ===")
    
    # 创建任务并立即开始执行
    task1 = asyncio.create_task(fetch_data("https://api1.example.com", 2))
    task2 = asyncio.create_task(fetch_data("https://api2.example.com", 1))
    
    # 等待任务完成并获取结果
    result1 = await task1
    result2 = await task2
    
    print(f"任务1结果: {result1}")
    print(f"任务2结果: {result2}")
    
    print(f"任务1状态: {task1.done()}")  # True
    print(f"任务2状态: {task2.done()}")  # True

# 运行事件循环
asyncio.run(main())

输出示例：

=== 基础任务创建 ===
开始获取 https://api1.example.com 的数据...
开始获取 https://api2.example.com 的数据...
完成获取 https://api2.example.com 的数据，耗时 1秒
完成获取 https://api1.example.com 的数据，耗时 2秒
任务1结果: {'url': 'https://api1.example.com', 'data': '来自 https://api1.example.com 的数据'}
任务2结果: {'url': 'https://api2.example.com', 'data': '来自 https://api2.example.com 的数据'}
任务1状态: True
任务2状态: True

示例2：使用 gather 并发执行多个任务

import asyncio

async def process_order(order_id, processing_time):
    """模拟订单处理"""
    print(f"开始处理订单 #{order_id}")
    await asyncio.sleep(processing_time)  # 模拟处理时间
    print(f"订单 #{order_id} 处理完成")
    return f"订单#{order_id}已完成"

async def main():
    print("=== 使用 gather 并发执行 ===")
    
    # 创建多个任务
    tasks = [
        asyncio.create_task(process_order(i, i * 0.5))
        for i in range(1, 4)
    ]
    
    # gather 会等待所有任务完成，并收集结果
    results = await asyncio.gather(*tasks)
    
    print(f"\n所有订单处理结果:")
    for result in results:
        print(f"  - {result}")

asyncio.run(main())

输出示例：

=== 使用 gather 并发执行 ===
开始处理订单 #1
开始处理订单 #2
开始处理订单 #3
订单 #1 处理完成
订单 #2 处理完成
订单 #3 处理完成

所有订单处理结果:
  - 订单#1已完成
  - 订单#2已完成
  - 订单#3已完成

示例3：任务的取消与异常处理

import asyncio

async def long_running_task():
    """一个可能长时间运行的任务"""
    print("长时间任务开始")
    try:
        # 模拟一个需要很长时间的操作
        await asyncio.sleep(10)
        return "任务正常完成"
    except asyncio.CancelledError:
        print("任务被取消！")
        # 可以选择重新抛出异常或返回默认值
        raise  # 重新抛出，调用者会看到 CancelledError

async def main():
    print("=== 任务取消示例 ===")
    
    # 创建任务
    task = asyncio.create_task(long_running_task())
    
    # 等待一小段时间后取消任务
    await asyncio.sleep(1)
    print("准备取消任务...")
    
    # 取消任务
    task.cancel()
    
    try:
        # 等待任务完成（包括被取消）
        result = await task
        print(f"任务结果: {result}")
    except asyncio.CancelledError:
        print("捕获到任务取消异常")
    
    print(f"任务是否被取消: {task.cancelled()}")  # True

asyncio.run(main())

示例4：使用 wait 灵活等待任务

import asyncio

async def api_call(call_id, timeout):
    """模拟API调用"""
    print(f"API调用 #{call_id} 开始")
    await asyncio.sleep(timeout)
    print(f"API调用 #{call_id} 完成")
    return f"API#{call_id}_响应"

async def main():
    print("=== 使用 wait 灵活等待 ===")
    
    # 创建多个任务
    tasks = [
        asyncio.create_task(api_call(i, i * 0.3))
        for i in range(1, 4)
    ]
    
    # wait 提供了更灵活的等待选项
    # return_when=asyncio.FIRST_COMPLETED: 当第一个任务完成时返回
    done, pending = await asyncio.wait(
        tasks, 
        return_when=asyncio.FIRST_COMPLETED
    )
    
    print(f"\n已完成的任务 ({len(done)} 个):")
    for task in done:
        print(f"  - 结果: {task.result()}")
    
    print(f"仍在运行的任务 ({len(pending)} 个):")
    # 可以选择等待剩余任务或取消它们
    for task in pending:
        task.cancel()

asyncio.run(main())

实践练习

练习1：基础任务管理（初级）

创建三个异步任务，分别模拟从三个不同的数据源获取数据（使用不同的延迟时间），然后按完成顺序输出结果。

要求：

每个任务返回数据源名称和随机生成的数据
使用 asyncio.gather() 等待所有任务完成
输出所有结果，显示实际执行顺序

练习2：任务取消与重试（中级）

实现一个带重试机制的异步网络请求模拟函数。

要求：

函数 fetch_with_retry(url, max_retries=3) 模拟可能失败的网络请求
使用 asyncio.wait_for() 设置超时（1秒）
请求失败或超时时进行重试
记录并输出每次重试的尝试次数和状态

练习3：任务编排（高级）

实现一个简单的"任务流水线"系统。

要求：

定义三个阶段：数据获取、数据处理、结果保存
每个阶段可以包含多个并发任务
使用 asyncio.create_task() 和 asyncio.gather() 编排任务
确保阶段按顺序执行（阶段内并发）
输出每个阶段的开始/结束时间和整体执行时间

常见错误

错误1：忘记等待任务完成

async def main():
    task = asyncio.create_task(some_coroutine())
    # 错误：没有 await task，程序可能提前结束
    # 正确做法：总是 await 你需要结果的任务
    result = await task  # 必须等待任务完成

错误2：在非协程中直接调用协程

async def my_coroutine():
    await asyncio.sleep(1)
    return "done"

# 错误：直接调用协程只是创建了协程对象，没有执行
result = my_coroutine()  # 返回的是协程对象，不是结果

# 正确做法：在事件循环中执行
async def main():
    result = await my_coroutine()  # 正确

错误3：重复 await 同一个任务

async def main():
    task = asyncio.create_task(some_coroutine())
    
    # 错误：多次 await 同一个已完成的任务
    result1 = await task  # 第一次 await 正常
    result2 = await task  # 第二次 await 会立即返回相同结果，但通常是逻辑错误
    
    # 如果需要多次获取结果，应该保存第一次的结果

错误4：忽略任务异常

async def risky_task():
    raise ValueError("出错了")

async def main():
    task = asyncio.create_task(risky_task())
    
    # 错误：不 await 或检查异常，程序可能静默失败
    # 正确做法：总是处理任务可能抛出的异常
    try:
        result = await task
    except ValueError as e:
        print(f"任务失败: {e}")

错误5：在错误上下文中运行事件循环

# 错误：在已有事件循环中调用 asyncio.run()
async def main():
    # ... 异步代码 ...

# 如果已经在事件循环中（如在Jupyter Notebook中），不要使用asyncio.run()
# 而是使用 await main() 或获取当前循环

小结

关键要点回顾

事件循环是核心：所有异步任务都在事件循环中调度执行，理解事件循环是掌握 asyncio 的关键。
任务是协程的增强：使用 asyncio.create_task() 将协程包装成任务，使其能够被事件循环调度，并可以并发执行。
并发不等于并行：任务在 I/O 等待时切换，但仍然是单线程执行。适合 I/O 密集型任务，而非 CPU 密集型任务。
常用模式：
- await task：等待单个任务完成
- await asyncio.gather(*tasks)：并发执行多个任务并收集所有结果
- await asyncio.wait(tasks)：更灵活的等待方式，支持不同的返回条件
任务生命周期管理：了解任务的创建、运行、完成（正常/异常/取消）状态，合理使用 cancel()、done()、result() 等方法。
错误处理：始终考虑任务可能失败的情况，使用 try/except 捕获 asyncio.CancelledError 和其他异常。

最佳实践

为长时间运行的任务设置超时
适当使用任务取消机制避免资源浪费
合理控制并发任务数量，避免过载
使用有意义的任务名称便于调试
在复杂场景下使用 asyncio.wait() 进行精细控制

下一步学习

掌握了任务和事件循环后，你已经具备了使用 asyncio 构建复杂异步程序的基础。下一课将学习网络编程中的 Socket 基础与 TCP 通信，这是许多网络应用的核心技术。异步编程在网络通信中尤为重要，因为网络 I/O 通常是主要的性能瓶颈。

练习编辑器

python