并发模型演进史（三）：协程——用户态的轻量级线程

传统模型：一个连接一个线程

  客户端 1 ──── 线程 1
  客户端 2 ──── 线程 2
  客户端 3 ──── 线程 3
  ...
  客户端 10000 ──── 线程 10000

问题：
  - 每个线程占用 1-8MB 栈空间
  - 10000 个线程 = 10-80GB 内存
  - 线程切换开销巨大
  - 操作系统调度器崩溃

线程切换（内核态）：
  1. 陷入内核（系统调用）
  2. 保存当前线程状态
  3. 调度器选择下一个线程
  4. 恢复下一个线程状态
  5. 返回用户态

  代价：~1000-10000 CPU 周期（取决于 OS 和硬件）

协程切换（用户态）：
  1. 保存当前协程状态（几个寄存器）
  2. 恢复下一个协程状态
  完了。

  代价：~100 CPU 周期（取决于实现）

线程 = 餐厅服务员
  - 由餐厅经理（OS）调度
  - 每个服务员有自己的工具车（栈空间）
  - 换人服务要走完整交接流程

协程 = 一个服务员分饰多角
  - 自己决定什么时候换角色
  - 共用一套工具
  - 换角色只需要换个帽子

普通函数：
  调用 → 执行完 → 返回
  中间不能停

协程函数：
  调用 → 执行一部分 → 暂停（yield）→ 被恢复 → 继续执行 → 暂停 → ... → 返回
  中间可以停很多次

# Python 生成器协程
def count_up():
    i = 0
    while True:
        yield i      # 暂停，返回 i
        i += 1       # 恢复后从这里继续

counter = count_up()
print(next(counter))  # 0
print(next(counter))  # 1
print(next(counter))  # 2

有栈协程：
  ┌─────────────┐
  │ 协程 A 的栈  │  ← 独立栈空间（通常 2-8KB）
  │  - 局部变量  │
  │  - 返回地址  │
  │  - 调用链    │
  └─────────────┘

  ┌─────────────┐
  │ 协程 B 的栈  │
  │  - 局部变量  │
  │  - 返回地址  │
  │  - 调用链    │
  └─────────────┘

代表：Go goroutine, Lua coroutine

无栈协程：
  async fn fetch_data() {
      let a = step1().await;    // 暂停点 1
      let b = step2(a).await;   // 暂停点 2
      process(a, b)
  }

  编译成状态机：
  ┌─────────────────────────────┐
  │ FetchDataFuture             │
  │  state: enum { S1, S2, S3 } │  ← 当前状态
  │  a: Option<A>               │  ← 跨 await 的变量
  │  b: Option<B>               │
  └─────────────────────────────┘

代表：Rust async/await, JavaScript Promise, Python asyncio

// 创建一个 goroutine，就这么简单
go func() {
    // 这里是另一个执行流
    fmt.Println("Hello from goroutine")
}()

1. 极小的初始栈（Go 1.4+ 为 2KB，可动态增长）
   普通线程：1-8MB（取决于 OS 配置）
   goroutine：2KB 起步（早期版本是 4KB/8KB）
   差距：数百到数千倍

2. 可以创建百万个
   func main() {
       for i := 0; i < 1_000_000; i++ {
           go func() {
               time.Sleep(time.Hour)
           }()
       }
   }
   // 用不了多少内存

3. M:N 调度
   M 个 goroutine 映射到 N 个 OS 线程
   Go runtime 负责调度

GMP 模型：
  G = Goroutine（协程）
  M = Machine（OS 线程）
  P = Processor（逻辑处理器）

  ┌─────────────────────────────────┐
  │           Go Runtime            │
  │  ┌─────┐  ┌─────┐  ┌─────┐     │
  │  │  P  │  │  P  │  │  P  │     │  ← P 的数量 = GOMAXPROCS
  │  └──┬──┘  └──┬──┘  └──┬──┘     │
  │     │        │        │         │
  │  ┌──┴──┐  ┌──┴──┐  ┌──┴──┐     │
  │  │  M  │  │  M  │  │  M  │     │  ← M 绑定到 OS 线程
  │  └─────┘  └─────┘  └─────┘     │
  └─────────────────────────────────┘

每个 P 有一个本地队列：
  P1: [G1, G2, G3, G4]
  P2: [G5, G6]
  P3: [G7, G8, G9]

  还有一个全局队列：
  Global: [G10, G11, G12, ...]

work stealing（工作窃取）：

  P1 的队列空了：[空]
  P2 的队列满：[G1, G2, G3, G4]

  P1 会从 P2 偷一半任务：
  P1: [G3, G4]
  P2: [G1, G2]

  负载自动均衡

系统调用处理：

  G1 要做系统调用（比如读文件）
    │
    ├─ M1 带着 G1 去做系统调用
    │
    └─ P1 找另一个 M2 继续执行其他 G
       不让 P1 闲着

  系统调用结束后：
    M1 尝试获取空闲的 P
    如果没有，G1 放入全局队列

// 这个 async 函数
async fn fetch_user(id: u64) -> User {
    let data = fetch_from_db(id).await;
    let user = parse_user(data).await;
    user
}

// 编译器转换成这样的状态机
enum FetchUserFuture {
    Start { id: u64 },
    WaitingDb { id: u64, fetch_future: DbFuture },
    WaitingParse { data: Data, parse_future: ParseFuture },
    Done,
}

impl Future for FetchUserFuture {
    type Output = User;

    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<User> {
        loop {
            match self.state {
                Start { id } => {
                    let future = fetch_from_db(id);
                    self.state = WaitingDb { id, fetch_future: future };
                }
                WaitingDb { fetch_future, .. } => {
                    match fetch_future.poll(cx) {
                        Poll::Pending => return Poll::Pending,
                        Poll::Ready(data) => {
                            let future = parse_user(data);
                            self.state = WaitingParse { data, parse_future: future };
                        }
                    }
                }
                // ...
            }
        }
    }
}

1. 零成本抽象
   async 函数编译后，和手写状态机一样高效
   没有运行时开销

2. 精确的内存占用
   编译器精确计算 Future 大小
   不需要预分配栈

3. 无 GC 配合所有权
   异步代码也受所有权规则保护
   编译期防止数据竞争

4. 可选的运行时
   std 不包含 async 运行时
   可以选择 tokio, async-std 等
   甚至可以在嵌入式环境用

// 一个 async 函数
async fn async_operation() -> i32 {
    42
}

// 调用它的函数也得是 async
async fn caller() -> i32 {
    async_operation().await
}

// 一路传染上去
async fn main_logic() -> i32 {
    caller().await
}

// 直到 main 或运行时边界
#[tokio::main]
async fn main() {
    let result = main_logic().await;
    println!("{}", result);
}

func main() {
    // 创建协程，用 go 关键字
    go doSomething()

    // channel 通信
    ch := make(chan int)
    go func() {
        ch <- 42
    }()
    result := <-ch
}

#[tokio::main]
async fn main() {
    // 创建异步任务
    let handle = tokio::spawn(async {
        do_something().await
    });

    // 等待完成
    let result = handle.await.unwrap();
}

import asyncio

async def main():
    # 创建任务
    task = asyncio.create_task(do_something())

    # 等待完成
    result = await task

asyncio.run(main())

async function main() {
    // Promise 和 async/await
    const result = await fetch('/api/data');
    const data = await result.json();
    return data;
}

main().then(console.log);

// 错误：在 async 中调用阻塞函数
async fn bad_example() {
    // 这会阻塞整个线程！
    std::thread::sleep(Duration::from_secs(1));
}

// 正确：使用异步版本
async fn good_example() {
    tokio::time::sleep(Duration::from_secs(1)).await;
}

tokio 默认配置：
  - 4 个 worker 线程（CPU 核心数）
  - 运行 10000 个 async 任务

如果一个任务阻塞了线程：
  - 只剩 3 个线程干活
  - 吞吐量下降 25%

如果 4 个任务同时阻塞：
  - 所有任务都卡住
  - 系统假死

// 你想在同步代码中调用 async 函数
fn sync_function() {
    // 这编译不过！
    // let result = async_function().await;

    // 需要用 block_on
    let result = tokio::runtime::Runtime::new()
        .unwrap()
        .block_on(async_function());
}

async fn transfer(from: &Account, to: &Account, amount: u64) {
    from.deduct(amount).await;
    // 如果这里被取消...
    to.add(amount).await;  // 这行不会执行
    // 钱就丢了！
}

线程的问题：
  - 创建成本高（MB 级栈空间）
  - 切换成本高（内核态切换）
  - 数量有限（几千个就很多了）

协程的解决方案：
  - 用户态切换，快 10-100 倍
  - 更小的栈或无栈，可创建百万个
  - 程序自己调度，更可控