2. 异步编程基础回顾：async/await、事件循环、协程与任务

在深入数据库迁移异步任务接口之前，必须牢固掌握 Python 异步编程的核心概念。本章将系统回顾 async/await 语法、事件循环（Event Loop）、协程（Coroutine）与任务（Task），为后续的异步数据库操作打下坚实基础。

2.1 核心概念对比

2.2 事件循环（Event Loop）——异步引擎

事件循环是异步编程的核心调度器。它不断重复以下步骤：

1. 轮询：检查是否有就绪的 I/O 事件（如网络数据到达、文件读取完成）。
2. 调度：将执行权交给等待该事件的协程。
3. 执行：协程运行直到遇到下一个 await 或完成。
4. 挂起：协程挂起时，事件循环记录其状态，继续处理其他任务。

2.3 协程（Coroutine）与 async/await 语法

协程是异步编程的基本执行单元。使用 async def 定义的函数返回一个协程对象，该对象必须被事件循环调度才能执行。

2.3.1 基础语法示例

import asyncio

# 定义一个协程函数
async def fetch_data(url: str) -> dict:
    print(f"开始请求: {url}")
    # 模拟网络 I/O 等待，await 挂起当前协程
    await asyncio.sleep(1)
    print(f"完成请求: {url}")
    return {"url": url, "status": 200}

# 直接调用协程函数返回协程对象，不会执行
coro = fetch_data("https://api.example.com/users")
print(type(coro))  # <class 'coroutine'>

# 正确执行方式：在事件循环中运行
async def main():
    result = await fetch_data("https://api.example.com/users")
    print(result)

asyncio.run(main())

2.4 任务（Task）——并发执行的关键

任务（asyncio.Task）是协程的并发包装器。通过 asyncio.create_task() 将协程包装为任务后，事件循环会自动调度它，无需显式 await 即可开始执行。

2.4.1 任务 vs 直接 await 对比

2.4.2 并发任务实战示例

import asyncio
import time

async def query_database(query_id: int) -> str:
    """模拟数据库查询"""
    await asyncio.sleep(1)  # 模拟 I/O 等待
    return f"Query-{query_id} result"

async def main():
    start = time.time()

    # 创建 5 个并发任务
    tasks = [
        asyncio.create_task(query_database(i))
        for i in range(5)
    ]

    # 等待所有任务完成，收集结果
    results = await asyncio.gather(*tasks)

    elapsed = time.time() - start
    print(f"耗时: {elapsed:.2f}s")  # 约 1 秒，而非 5 秒
    print(f"结果: {results}")

asyncio.run(main())

2.5 异步上下文管理器与异步迭代器

在数据库迁移场景中，异步上下文管理器（async with）和异步迭代器（async for）是管理数据库连接和游标的核心工具。

2.5.1 异步上下文管理器

class AsyncDatabaseConnection:
    async def __aenter__(self):
        print("打开数据库连接")
        self.conn = await create_connection()
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("关闭数据库连接")
        await self.conn.close()

# 使用
async with AsyncDatabaseConnection() as conn:
    result = await conn.execute("SELECT 1")

2.5.2 异步迭代器

class AsyncResultSet:
    def __init__(self, data):
        self._data = data
        self._index = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self._index >= len(self._data):
            raise StopAsyncIteration
        # 模拟异步获取下一行
        await asyncio.sleep(0.1)
        value = self._data[self._index]
        self._index += 1
        return value

# 使用
async for row in AsyncResultSet([1, 2, 3]):
    print(row)

2.6 本章小结