Python 异步编程完整解析：从事件循环到协程的底层实现

# 事件循环的伪代码
def event_loop():
    while True:
        # 1. 检查定时器（TimerHandle）
        fire_due_timers()

        # 2. 执行 I/O 事件（通过 epoll/select）
        for fd, event in poll(events, timeout):
            if event & READABLE:
                schedule_callback(read_handlers[fd])
            if event & WRITABLE:
                schedule_callback(write_handlers[fd])

        # 3. 执行准备就绪的回调
        while ready_queue:
            callback = ready_queue.pop(0)
            callback()

        # 4. 执行空闲任务（idle callbacks）

flowchart TD
    subgraph "事件循环核心流程"
        A[开始新迭代] --> B[计算最小超时时间]
        B --> C[调用 select/epoll 等待事件]
        C --> D{有事件就绪?}
        D -->|是| E[读取事件列表]
        D -->|否| F[超时返回]
        E --> G[执行 I/O 回调]
        F --> H[执行到期定时器]
        G --> H
        H --> I[执行回调队列]
        I --> J{还有待处理事件?}
        J -->|是| C
        J -->|否| A
    end

async def hello():
    return "Hello World"

print(hello())        # 
print(type(hello()))  # 

# 协程的简化实现
class Coroutine:
    def __await__(self):
        return self._iter_impl().__await__()

    def send(self, value):
        # 恢复执行，直到下一个 await 或 return
        pass

    def throw(self, exception):
        # 在协程内部抛出异常
        pass

async def fetch_data():
    # 1. 创建协程，但不执行
    result = await async_io_operation()  
    # 2. 挂起当前协程，yield 控制权到事件循环
    # 3. 当 async_io_operation() 完成时，事件循环 send 结果回来
    # 4. 恢复执行
    return result

sequenceDiagram
    participant EL as 事件循环
    participant C1 as 协程A
    participant C2 as 协程B
    participant IO as IO 操作

    EL->>C1: send(None) 启动
    C1->>C1: 执行到第一个 await
    C1->>IO: 发起 IO 操作
    C1->>EL: 挂起（yield 控制权）
    EL->>C2: send(None) 启动
    C2->>C2: 执行代码
    C2->>IO: 发起 IO 操作
    C2->>EL: 挂起
    Note over EL: 等待 IO 事件...
    IO-->>EL: 数据就绪
    EL->>C1: send(data) 恢复
    C1->>C1: 处理结果
    C1->>EL: 完成

class Task(Future):
    def __init__(self, coro, *, loop=None, name=None):
        super().__init__(loop=loop)
        self._coro = coro           # 包装的协程
        self._name = name
        self._loop.call_soon(self.__step)

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                result = coro.send(None)  # 推进协程
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            # 协程完成
            self.set_result(exc.value)
        except Exception as exc:
            self.set_exception(exc)
        else:
            # 协程挂起，返回了一个 Future
            if isinstance(result, Future):
                result.add_done_callback(self.__wakeup)

flowchart TD
    subgraph "任务调度"
        T1[Task 提交] --> Q[就绪队列]
        T2[IO 完成] --> Q
        T3[定时器到期] --> Q
        Q --> S[__step 依次执行]
        S --> R{协程返回?}
        R -->|await Future| F[Future 未完成]
        R -->|StopIteration| D[任务完成]
        F --> W[注册回调: 完成后重新入队]
    end

async def gather(*coros, return_exceptions=False):
    # 为每个协程创建 Task（隐式调度）
    tasks = [ensure_future(c) for c in coros]

    if not tasks:
        return []

    # 内部使用 _GatheringFuture 等待所有任务
    outer = _GatheringFuture(tasks, loop=loop)
    for i, t in enumerate(tasks):
        t.add_done_callback(partial(outer._done_callback, i))

    return await outer  # 挂起直到所有 task 完成

class EchoProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        # TCP 数据到达回调
        self.transport.write(data)  # 回写

    def connection_lost(self, exc):
        # 连接关闭
        pass

async def tcp_echo_client():
    reader, writer = await asyncio.open_connection(
        'localhost', 8888)

    writer.write(b'Hello')
    await writer.drain()  # 等待发送缓冲区清空

    data = await reader.read(100)
    writer.close()
    await writer.wait_closed()

async def worker(lock, name):
    async with lock:  # 异步上下文管理器
        print(f"{name} 获取了锁")
        await asyncio.sleep(1)

# 同步版本
def sync_fetch_all(urls):
    results = []
    for url in urls:
        resp = requests.get(url)  # 阻塞 100ms
        results.append(resp.text)
    return results
# 100 个 URL → 10 秒

# 异步版本
async def async_fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)
# 100 个 URL → ~100ms（并发时）

flowchart LR
    subgraph "同步模型"
        S1[请求1 100ms] --> S2[请求2 100ms] --> S3[请求3 100ms]
    end
    subgraph "异步模型"
        A1[请求1 100ms] -.- A2[请求2 100ms]
        A2 -.- A3[请求3 100ms]
        A1 -.- T[总时长 ≈100ms]
        A3 -.- T
    end