Python 异步架构详解：事件循环(event loops)、任务(tasks)、协程（coroutines）与 Future

简介

Asyncio 是 Python 中用于编写并发代码的强大库。但它的内部工作机制究竟是什么？在本文中，我们将深入探讨 asyncio 的三个核心组件：事件循环、任务和 Future。在本文结束时，你不仅会知道如何使用它们，还会理解它们是如何协同工作的。

---

1. 事件循环：核心调度器

事件循环是所有 asyncio 应用的核心。它负责跟踪多个任务，并在它们之间切换，让它们看似并行运行。

简单来说，事件循环就像一个无限循环，不断检查可执行的任务、执行它们，并处理已完成的结果。

它是如何工作的？

1. 事件循环维护一个待执行任务的列表
2. 当任务执行到 await 时，它会将控制权交还给事件循环
3. 事件循环接着运行下一个可执行的任务
4. 当被等待的任务准备好时，事件循环会恢复它

代码示例

import asyncio

async def display_message():
    print("开始")
    await asyncio.sleep(1)
    print("结束")

asyncio.run(display_message())

asyncio.run(main()) 在底层会：

• 创建一个新的事件循环
• 将 main() 作为任务运行
• 运行循环直到任务完成
• 关闭循环

---

2. 任务：并发执行单元

任务是协程的包装器，允许它们并发运行。当你将协程作为任务创建时，它会被调度到事件循环上，并独立于其他任务运行。

创建任务

使用 asyncio.create_task() 来创建任务。

代码示例

import asyncio

async def background_work():
    await asyncio.sleep(1)
    print("后台工作完成")

async def main():
    task = asyncio.create_task(background_work())
    print("任务已创建并调度")
    await task

asyncio.run(main())

注意：如果你不在 main() 中 await 任务，程序可能会在任务完成之前就结束。

---

3. 并发运行多个任务

你可以同时运行多个任务，并等待所有任务完成。

代码示例

import asyncio

async def fetch_user_data():
    await asyncio.sleep(2)
    return "用户数据已获取"

async def fetch_product_data():
    await asyncio.sleep(1)
    return "产品数据已获取"

async def main():
    user_task = asyncio.create_task(fetch_user_data())
    product_task = asyncio.create_task(fetch_product_data())
    
    user_result = await user_task
    product_result = await product_task
    print(user_result, product_result)

asyncio.run(main())

即使 fetch_product_data 在 1 秒后完成，程序也会等待 fetch_user_data 完成，因为我们在等待两个任务。

---

4. Future：占位符对象

Future 是一个低层级对象，代表一个尚未完成的异步操作的结果。它类似于 JavaScript 中的 Promise。

你可以将 Future 视为一个容器，它会在某个时刻存放一个值。任务（Task）实际上是 Future 的子类。

手动使用 Future

通常你不需要直接使用 Future，因为任务和 async/await 已经足够。但理解 Future 有助于理解底层机制。

代码示例

import asyncio

async def assign_future_value(future):
    await asyncio.sleep(1)
    future.set_result("值已设置")

async def main():
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    
    await asyncio.create_task(assign_future_value(future))
    result = await future
    print(result)

asyncio.run(main())

在这个示例中：

• 我们手动创建了一个 Future 对象
• 一个任务在 1 秒后为 Future 设置值
• 我们等待 Future 解析并打印结果

---

5. 任务与 Future 的区别

特性	任务（Task）	Future
用途	包装并调度协程	表示一个待定的结果
创建方式	asyncio.create_task()	loop.create_future()
是否可等待	是	是
是否主动执行	是（被调度到事件循环）	否（需要手动设置结果）
日常使用频率	高	低

---

6. 错误处理

异步代码中的异常不会自动传播，除非你等待该任务或 Future。

代码示例

import asyncio

async def failing_operation():
    await asyncio.sleep(1)
    raise ValueError("出错了")

async def main():
    task = asyncio.create_task(failing_operation())
    
    try:
        await task
    except ValueError as e:
        print(f"捕获到异常: {e}")

asyncio.run(main())

如果你不等待任务，异常将静默丢失（只会记录到日志）。

---

7. 超时处理

你可以使用 asyncio.wait_for() 为异步操作设置超时。

代码示例

import asyncio

async def slow_operation():
    await asyncio.sleep(3)
    return "完成"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时了")

asyncio.run(main())

---

8. 同时等待多个任务

asyncio.gather() 让你可以同时运行多个可等待对象并收集结果。

代码示例

import asyncio

async def compute_square(number):
    await asyncio.sleep(0.5)
    return number * number

async def main():
    results = await asyncio.gather(
        compute_square(2),
        compute_square(3),
        compute_square(4)
    )
    print(f"平方结果: {results}")

asyncio.run(main())

如果某个任务抛出异常，gather() 会默认立即抛出该异常。

---

9. 完成时处理任务

asyncio.as_completed() 会在任务完成时逐个产生结果，而不是等待所有任务完成。

代码示例

import asyncio

async def slow_task(delay, name):
    await asyncio.sleep(delay)
    return f"任务 {name} 完成"

async def main():
    tasks = [
        slow_task(3, "A"),
        slow_task(1, "B"),
        slow_task(2, "C")
    ]
    
    for completed in asyncio.as_completed(tasks):
        result = await completed
        print(result)

asyncio.run(main())

输出顺序将是：

任务 B 完成
任务 C 完成
任务 A 完成

---

10. 事件循环的替代实现

除了标准的 asyncio 事件循环，还有其他实现：

• uvloop：基于 libuv 的更快实现，Node.js 也在使用
• proactor事件循环：Windows 上用于处理文件的特殊循环

你可以通过以下方式使用 uvloop：

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
asyncio.run(main())

---

总结

概念	一句话解释
事件循环	管理和调度所有异步任务的无限循环
协程	可以暂停和恢复的 async def 函数
任务	包装协程并允许并发执行的 Future
Future	代表将来某个时刻会得到结果的占位符

核心要点

1. 事件循环是 asyncio 的核心，负责任务调度
2. 使用 create_task() 并发运行任务
3. Future 是低层级的占位符，Task 是 Future 的子类
4. 始终等待你的任务，否则异常可能会丢失
5. 使用 gather()、as_completed() 和 wait_for() 处理常见的并发模式

理解这些组件将帮助你在 Python 中编写更高效、更易调试的异步代码。