Python中的协程与异步编程底层实现（asyncio库详解）

1. 协程的基本概念

协程（Coroutine） 是一种用户态的轻量级线程，由用户控制调度而非操作系统。与多线程不同，协程在单线程内通过任务切换实现并发，避免了线程切换的开销和锁的复杂性。

• 关键特性：
  • 执行过程中可暂停（await）和恢复。
  • 通过事件循环（Event Loop）管理多个协程的调度。
  • 适用于I/O密集型任务（如网络请求、文件读写）。

2. 协程的创建方式

Python中协程可通过以下方式定义：

（1）async def 函数

async def my_coroutine():
    print("Start")
    await asyncio.sleep(1)  # 模拟I/O操作
    print("End")

• 使用 async def 定义的函数返回一个协程对象，但不会立即执行。
• 必须通过事件循环或 await 触发执行。

（2）@asyncio.coroutine 装饰器（旧版，已不推荐）

@asyncio.coroutine
def old_style_coroutine():
    yield from asyncio.sleep(1)

3. 事件循环（Event Loop）

事件循环是协程调度的核心，负责监听事件、管理任务队列和回调。

import asyncio

# 获取事件循环
loop = asyncio.get_event_loop()

# 运行协程
loop.run_until_complete(my_coroutine())

• 任务（Task）：
  • 对协程的进一步封装，表示可调度的执行单元。
  • 通过 asyncio.create_task() 或 loop.create_task() 创建。

4. await 关键字的作用

• await 用于暂停当前协程，将控制权交还给事件循环，直到等待的对象（如其他协程、Task、Future）完成。
• 可等待对象（Awaitable） 包括：
  • 协程（由 async def 定义）。
  • asyncio.Task。
  • asyncio.Future（底层异步操作的结果容器）。

示例：

async def fetch_data():
    await asyncio.sleep(2)  # 模拟网络请求
    return "Data"

async def main():
    result = await fetch_data()  # 等待fetch_data完成
    print(result)

5. 并发执行多个协程

使用 asyncio.gather() 或 asyncio.create_task() 实现并发：

async def task1():
    await asyncio.sleep(1)
    return "Task1"

async def task2():
    await asyncio.sleep(2)
    return "Task2"

async def main():
    # 方式1：gather统一调度
    results = await asyncio.gather(task1(), task2())
    print(results)  # ['Task1', 'Task2']

    # 方式2：单独创建Task
    t1 = asyncio.create_task(task1())
    t2 = asyncio.create_task(task2())
    result1 = await t1
    result2 = await t2

• gather() 会等待所有任务完成，并返回结果列表。
• 直接创建 Task 可更灵活控制任务生命周期。

6. 底层实现：Future与事件循环调度

• Future：
  • 是一个低级对象，代表异步操作的最终结果。
  • 协程内部通过 await 隐式将协程转换为 Future 对象。
• 事件循环的工作流程：
  1. 维护一个任务队列（Ready Queue）和待调度队列（如I/O等待队列）。
  2. 循环检查任务状态，当某个I/O操作完成时，将对应的Future标记为完成，并移回就绪队列。
  3. 通过回调机制唤醒等待该Future的协程。

示例模拟Future：

async def slow_operation():
    future = asyncio.Future()
    # 模拟异步操作完成后设置结果
    asyncio.get_event_loop().call_later(2, future.set_result, "Done")
    return await future

7. 异步上下文管理器与异步迭代器

• 异步上下文管理器（async with）：

  async def async_context_manager():
      async with some_async_resource() as resource:
          data = await resource.fetch()
  

• 异步迭代器（async for）：

  async for item in async_iterator:
      await process(item)
  

8. 实际应用：异步HTTP请求

使用 aiohttp 库实现并发网络请求：

import aiohttp
import asyncio

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["http://example.com", "http://example.org"]
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)

总结

• 协程通过 async/await 语法实现单线程内并发。
• 事件循环是调度核心，Future是底层结果容器。
• 适用场景：高并发I/O操作，避免多线程的锁和切换开销。
• 注意：异步代码中避免使用阻塞操作（如 time.sleep），需用异步替代（asyncio.sleep）。