异步与协程精讲 / 第6章：协程 —— 用户态的并发原语

第6章：协程 —— 用户态的并发原语

6.1 什么是协程？

协程（Coroutine）是一种可以暂停和恢复执行的程序组件。与函数只能从头到尾执行一次不同，协程可以在执行过程中**让出（yield）控制权，之后从暂停的地方恢复（resume）**执行。

类比：协程就像一本可以随时插入书签、合上、下次从书签处继续阅读的书。而普通函数就像一张收据——要么一口气读完，要么丢掉从头来。

协程 vs 函数

历史

协程的概念比线程还要古老：

• 1958年：Melvin Conway 首次提出"协程"一词，用于编译器设计
• 1963年：Simula 语言中使用协程实现离散事件模拟
• 1975年：Donald Knuth 在 TAOCP 中描述协程
• 2003年：C# 2.0 引入 yield return（迭代器协程）
• 2012年：Go 语言以 goroutine 普及协程
• 2015年：Python 3.5 引入 async/await（原生协程）
• 2019年：Rust 引入 async/await
• 2020年：C++20 引入协程

6.2 协程的基本操作

创建、让出、恢复

# Python — 最简单的协程演示
def simple_coroutine():
    print("→ 协程启动")
    yield 1                    # 让出控制权，返回值 1
    print("→ 协程恢复")
    yield 2                    # 再次让出
    print("→ 协程结束")

# 创建协程（不会立即执行）
coro = simple_coroutine()

# 恢复执行
print(next(coro))     # → 协程启动 → 输出 1
print(next(coro))     # → 协程恢复 → 输出 2
# next(coro)          # → 协程结束 → StopIteration

双向通信（Send）

def accumulator():
    total = 0
    while True:
        value = yield total   # yield 当前总和，接收新值
        if value is None:
            break
        total += value

acc = accumulator()
next(acc)               # 启动协程（必须先 next/send(None)）
print(acc.send(10))     # 发送 10，返回 10
print(acc.send(20))     # 发送 20，返回 30
print(acc.send(30))     # 发送 30，返回 60

状态机视图

协程状态转换：

  ┌─────┐
  │创建  │  coro = my_coroutine()
  └──┬──┘
     │
     ▼
  ┌─────┐  next()/send()  ┌──────┐
  │挂起  │ ──────────────► │执行中 │
  │SUSPENDED│              │RUNNING│
  └──┬──┘  ◄────────────── └──┬──┘
     │      yield              │
     │                         │
     ▼                    StopIteration
  ┌─────┐                   │
  │完成  │ ◄─────────────────┘
  │CLOSED│
  └─────┘

6.3 对称协程 vs 非对称协程

非对称协程（Asymmetric Coroutine）

也称为半对称协程（Semi-coroutine），是现代语言中最常见的形式。

特征：

• 协程只能将控制权归还给它的调用者
• 存在明确的"调用"和"返回"关系
• 类似函数调用，但可以暂停

# Python 的生成器就是非对称协程
def generator():
    yield 1    # 控制权归还给调用方
    yield 2    # 控制权归还给调用方

for val in generator():
    print(val)

代表语言：Python、JavaScript、Rust、C#、C++

对称协程（Symmetric Coroutine）

特征：

• 协程可以直接将控制权转移给任意其他协程
• 没有固定的调用关系
• 使用 transfer(target) 操作

-- Lua 的协程是对称协程
local co1, co2

co1 = coroutine.create(function()
    print("co1: 开始")
    coroutine.transfer(co2)   -- 直接转移到 co2
    print("co1: 恢复")
end)

co2 = coroutine.create(function()
    print("co2: 执行")
    coroutine.transfer(co1)   -- 直接转移到 co1
end)

coroutine.transfer(co1)  -- 从主线程转移到 co1

对比

6.4 有栈协程 vs 无栈协程

这是协程实现方式的根本分类。

有栈协程（Stackful Coroutine）

每个协程拥有独立的调用栈。

协程 A 的栈        协程 B 的栈       主线程的栈
┌──────────┐     ┌──────────┐     ┌──────────┐
│ func_c() │     │ func_f() │     │ main()   │
│ func_b() │     │ func_e() │     │          │
│ func_a() │     │ func_d() │     │          │
└──────────┘     └──────────┘     └──────────┘

特点：

代表：Go（goroutine）、Erlang、Lua、Boost.Fiber、Ruby Fiber

无栈协程（Stackless Coroutine）

协程状态保存在一个状态对象中，没有独立调用栈。

协程状态对象：
┌─────────────────────────┐
│ 当前挂起点（suspend point）│
│ 局部变量 a = 10          │
│ 局部变量 b = "hello"     │
│ 上下文数据 ...            │
└─────────────────────────┘

特点：

代表：Rust（async/await）、Python（async/await）、C++20、C#、JavaScript

关键区别

# 有栈协程 — 可以在任意深度让出
def deep_function():
    result = some_calculation()
    yield result           # ✅ 可以在这里让出
    return more_work()

# 无栈协程 — 只能在标记处让出
async def deep_function():
    result = await some_calculation()   # ✅ await 标记
    # 普通函数调用内部即使有异步操作，也无法从这里让出
    return more_work()

6.5 协程 vs 线程

这是初学者最常问的问题。

本质区别

抢占式 vs 协作式

抢占式调度（线程）：
  线程A: ████░░░░████░░░░████    ← OS 随时中断
  线程B: ░░░░████░░░░████░░░░    ← OS 随时切换

协作式调度（协程）：
  协程A: ████──────████──────    ← 主动让出
  协程B: ──────████──────████    ← 在让出点切换

注意：Go 的 goroutine 是协作式调度，但从 Go 1.14 开始引入了基于信号的异步抢占，避免了长时间运行的 goroutine 阻塞调度器。

内存对比

假设同时运行 10,000 个并发任务：

6.6 协程的调度模型

对称调度（Symmetric Scheduling）

调度器
  │
  ├──► 协程 A ──transfer──► 协程 B ──transfer──► 协程 C
  │         ◄─transfer───────────────transfer──┘
  │
  └──► 协程自己决定转移给谁

非对称调度（Asymmetric Scheduling）

调度器（事件循环）
  │
  ├──► 运行协程 A
  │      │ yield
  │      ▼
  ├──► 运行协程 B
  │      │ yield
  │      ▼
  ├──► 运行协程 C
  │      │ yield
  │      ▼
  └──► 回到调度器，选择下一个就绪的协程

N:M 调度模型

Go 的 GMP 模型是 N:M 调度的经典实现：

N 个 goroutine → M 个 OS 线程

G (Goroutine)   : 用户级协程，数量可达数百万
M (Machine)     : OS 线程，数量通常等于 CPU 核数
P (Processor)   : 逻辑处理器，持有运行队列

┌──────────────────────────────────────────┐
│              Go Scheduler                │
│                                          │
│  ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐       │
│  │ P0  │ │ P1  │ │ P2  │ │ P3  │       │
│  │[G,G]│ │[G]  │ │[G,G]│ │[G]  │       │
│  └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘       │
│     │       │       │       │            │
│     ▼       ▼       ▼       ▼            │
│  ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐       │
│  │ M0  │ │ M1  │ │ M2  │ │ M3  │       │
│  │(线程)│ │(线程)│ │(线程)│ │(线程)│       │
│  └─────┘ └─────┘ └─────┘ └─────┘       │
└──────────────────────────────────────────┘

6.7 协程的组合与通信

通信方式一：通道（Channel）

// Go — Channel 通信
func producer(ch chan<- int) {
    for i := 0; i < 10; i++ {
        ch <- i
    }
    close(ch)
}

func consumer(ch <-chan int) {
    for v := range ch {
        fmt.Println(v)
    }
}

func main() {
    ch := make(chan int, 5) // 带缓冲的通道
    go producer(ch)
    go consumer(ch)
}

通信方式二：Future/Promise

# Python — 使用 Future 通信
async def producer(future: asyncio.Future):
    await asyncio.sleep(1)
    future.set_result("数据已准备好")

async def consumer(future: asyncio.Future):
    result = await future  # 等待 producer 设置结果
    print(f"收到: {result}")

通信方式三：共享状态（需要同步）

// Rust — Arc<Mutex<T>> 共享状态
use std::sync::{Arc, Mutex};

let shared = Arc::new(Mutex::new(Vec::new()));

let shared_clone = shared.clone();
tokio::spawn(async move {
    let mut data = shared_clone.lock().unwrap();
    data.push(42);
});

6.8 业务场景：协程在 Web 框架中的应用

场景

一个高并发 API 服务器，每个请求需要查询数据库、调用外部 API、缓存结果。

# Python — FastAPI + 协程
from fastapi import FastAPI
import httpx
import asyncio

app = FastAPI()

@app.get("/dashboard/{user_id}")
async def get_dashboard(user_id: int):
    # 并发执行三个独立操作
    async with httpx.AsyncClient() as client:
        user_task = client.get(f"https://api/users/{user_id}")
        orders_task = client.get(f"https://api/orders?user={user_id}")
        recommendations_task = client.get(f"https://api/recs/{user_id}")

        user_resp, orders_resp, recs_resp = await asyncio.gather(
            user_task, orders_task, recommendations_task
        )

    return {
        "user": user_resp.json(),
        "orders": orders_resp.json(),
        "recommendations": recs_resp.json(),
    }

6.9 本章小结

下一章预告：我们将深入 Go 语言的 goroutine，看看它是如何用 CSP 模型实现简洁高效的并发编程。

扩展阅读

• A Tutorial Introduction to Coroutine Theory — Lewis Baker
• Coroutines in C — Simon Tatham
• Stackful Coroutine vs Stackless Coroutine — C++ 提案
• Go scheduler — Daniel Morsing
• Coroutines: Basics — Joshua Haberman