并发与异步：先划清所有权和取消边界

课程概览 · 第 18 章
上一篇：智能指针：让资源归属可组合
下一篇：Unsafe：把不安全性封装在最小边界

线程可以并行执行计算，异步让任务在等待时交还执行权。两者都应先明确数据归谁、谁能修改、任务如何结束，再决定如何启动。本章先用标准库演示线程边界，再用独立 Tokio 项目区分完成、超时、协作关闭与 abort 取消；不要求与其他章节拼成同一个应用。

普通与 scoped 线程、Future 丢弃、JoinHandle detach 和短同步锁的边界 — 图：普通 spawn 与 scope 的寿命约束不同；丢弃 Future 不回滚副作用，丢弃 JoinHandle 也不取消任务。

前置知识与学习目标

需要所有权、闭包、Result，以及第 17 章的 Arc 与内部可变性。学完本章应能：

用 scope 借用数据并检查 join 结果。
通过通道移交所有权，或以短临界区保护真实共享状态。
根据 Send/Sync 和跨 await 存活状态定位编译错误。
使用有界队列、timeout 与明确同步协议区分关闭和取消。

并行计算与异步等待

线程有独立栈与调度成本，可以在多个核上并行。async fn 返回 Future 状态机，不自动创建线程或开始执行；运行时轮询它，await 的对象未就绪时才可能暂停当前任务。Future 被唤醒后等待再次轮询，并非运行时只轮询已就绪对象。长时间同步计算会占住执行器线程，async 不会自动把它改成并行计算。

少量 CPU 工作可使用线程；异步 I/O 工作需要合适的执行器及 I/O 驱动。必要的同步阻塞可放入 spawn_blocking，但它也不是任意 CPU 任务的无限扩容方案，且已经开始执行的 blocking 任务不能按普通 async 任务的方式 abort。

scoped 线程与 join

thread::scope 保证作用域结束前派生线程结束，因此能借用当前函数的切片。普通 thread::spawn 要求闭包及返回值满足 Send + 'static；'static 表示不包含短命借用，不表示对象必须永远存活。

独立示例（仅标准库；保存为 src/main.rs，执行 cargo +stable run）：

use std::thread;

fn sum_parallel(values: &[i64]) -> i64 {
  thread::scope(|scope| {
    let (left, right) = values.split_at(values.len() / 2);
    let left_task = scope.spawn(|| left.iter().sum::<i64>());
    let right_sum: i64 = right.iter().sum();
    left_task.join().expect("worker panicked") + right_sum
  })
}

fn main() {
  let values: Vec<i64> = (1..=1000).collect();
  assert_eq!(sum_parallel(&values), 500_500);
  assert_eq!(sum_parallel(&[]), 0);
  assert_eq!(sum_parallel(&[7]), 7);
}

join 的 Result 暴露派生线程 panic，不能只启动而忽略结果。本例只展示所有权和空切片边界，不保证短输入比顺序求和更快。

消息传递与发送端关闭

send(String) 移动字符串；发送者之后不能继续访问原变量。克隆的是发送端句柄，不是消息。标准库 mpsc 的接收迭代在所有发送端销毁且缓冲消息耗尽后结束。

独立示例（仅标准库；src/main.rs，cargo +stable run）：

use std::sync::mpsc;
use std::thread;

fn main() {
  let (tx, rx) = mpsc::channel::<String>();
  let mut handles = Vec::new();
  for id in 0..3 {
    let tx = tx.clone();
    handles.push(thread::spawn(move || {
      tx.send(format!("worker-{id}"))
        .expect("receiver is alive");
    }));
  }
  drop(tx);
  let mut reports: Vec<_> = rx.into_iter().collect();
  for handle in handles {
    handle.join().expect("worker panicked");
  }
  reports.sort();
  assert_eq!(reports, vec!["worker-0", "worker-1", "worker-2"]);
}

输出到达顺序由调度决定，排序只是让断言稳定。保留一个无用发送端却等待接收迭代结束，会一直等待；关闭是端点生命周期协议，不是线程被强杀。

Arc<Mutex>：确实共享时再加锁

多个线程必须修改同一个统计值时，Arc 共享归属，Mutex 保护访问。guard 离开作用域释放锁，不是绑定后的分号释放锁。

独立示例（仅标准库；src/main.rs，cargo +stable run）：

use std::sync::{Arc, Mutex};
use std::thread;

#[derive(Default)]
struct Counters {
  completed: u64,
  failed: u64,
}

fn main() {
  let shared = Arc::new(Mutex::new(Counters::default()));
  let mut handles = Vec::new();
  for _ in 0..8 {
    let shared = Arc::clone(&shared);
    handles.push(thread::spawn(move || {
      let mut counters = shared.lock().expect("poisoned lock");
      counters.completed += 1;
      if counters.completed % 4 == 0 {
        counters.failed += 1;
      }
    }));
  }
  for handle in handles {
    handle.join().expect("worker panicked");
  }
  let counters = shared.lock().expect("poisoned lock");
  assert_eq!(counters.completed, 8);
  assert_eq!(counters.failed, 2);
}

短临界区不要做 I/O 或调用未知代码。标准库锁中毒是提示可能破坏了不变量，不是不可恢复的永久错误；恢复需检查状态，必要时使用 clear_poison，不应无条件忽略 PoisonError。本例把中毒视为违反程序不变量。

Send 与 Sync

Send 表示值能跨线程转移；Sync 表示 &T 能跨线程共享，等价于 &T: Send。通常由字段自动推导；手写 unsafe impl 是安全合同，不是消除报错的技巧。

故意失败：编译错误（仅标准库；独立 src/main.rs。Rc 不实现 Send，闭包不能传给普通 spawn）：

use std::rc::Rc;
use std::thread;

fn main() {
  let shared = Rc::new(vec![1, 2, 3]);
  let handle = thread::spawn(move || shared.len());
  let _ = handle.join();
}

独立示例（仅标准库；src/main.rs，cargo +stable run）：

use std::sync::Arc;
use std::thread;

fn main() {
  let shared = Arc::new(vec![1, 2, 3]);
  let worker = Arc::clone(&shared);
  let handle = thread::spawn(move || worker.len());
  assert_eq!(handle.join().expect("worker panicked"), 3);
  assert_eq!(shared.len(), 3);
}

Tokio 的多线程 spawn 同样要求 Future 为 Send + 'static，关键是哪些状态会跨 await 保留。标准库 MutexGuard 不 Send，把它保留过 await 可能使整个 Future 不满足 spawn 的约束，但不能推广为“所有 guard 跨 await 都编译失败”；Tokio 的异步 MutexGuard 可以跨 await。短内存临界区优先用标准库锁并在 await 前结束作用域，确需等待期间持锁时再评估异步锁。

Tokio：运行时、有界队列与超时

有界通道满时 send().await 等待容量，形成背压；不能预测“第三次 send 必然等待”，因为 worker 可能已经取走消息。下面只有一个 worker，按队列顺序处理四个 Job，并不让四项作业并行。Job 只有标题和 Fast/Slow 分类：Fast 立即就绪，Slow 永久 pending，分类结果不依赖固定睡眠能否及时被调度。

多文件示例（独立项目 task-workers；在项目根执行下述命令。第三方依赖仅 Tokio，完整 manifest 与源码如下，不沿用其他章定义）：

cargo +stable new --bin --edition 2024 task-workers
cd task-workers
# 用下述完整内容替换 Cargo.toml 与 src/main.rs
cargo +stable run

多文件示例，文件 task-workers/Cargo.toml（完整内容；执行目录 task-workers）：

[package]
name = "task-workers"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] }

多文件示例，文件 task-workers/src/main.rs（完整内容；依赖上面的 manifest；包含处理、协作关闭与独立取消协议）：

use std::future::pending;
use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
use tokio::time::timeout;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum JobKind {
  Fast,
  Slow,
}

#[derive(Debug)]
struct Job {
  title: String,
  kind: JobKind,
}

fn make_jobs() -> Vec<Job> {
  ["索引重建", "缓存预热", "报表导出", "慢备份"]
    .into_iter()
    .enumerate()
    .map(|(index, title)| Job {
      title: title.to_string(),
      kind: if index == 3 { JobKind::Slow } else { JobKind::Fast },
    })
    .collect()
}

async fn process(job: &Job) -> JobKind {
  match job.kind {
    JobKind::Fast => JobKind::Fast,
    JobKind::Slow => pending::<JobKind>().await,
  }
}

#[tokio::main]
async fn main() {
  let (tx, mut rx) = mpsc::channel::<Job>(2);
  let producer = tokio::spawn(async move {
    for job in make_jobs() {
      tx.send(job).await.expect("worker is alive");
    }
    // 最后一个发送端销毁，worker 排空队列后观察到 None。
  });

  let mut completed = 0usize;
  let mut timed_out = 0usize;
  while let Some(job) = rx.recv().await {
    match timeout(Duration::from_millis(50), process(&job)).await {
      Ok(kind) => {
        assert_eq!(kind, JobKind::Fast);
        println!("[worker] 完成: {}", job.title);
        completed += 1;
      }
      Err(_) => {
        assert_eq!(job.kind, JobKind::Slow);
        println!("[worker] 超时: {}", job.title);
        timed_out += 1;
      }
    }
  }
  producer.await.expect("producer panicked");
  println!("[worker] 通道关闭，worker 退出");
  assert_eq!(completed, 3);
  assert_eq!(timed_out, 1);

  // 独立协作关闭例：关闭并排空通道，recv 返回 None。
  let (close_tx, mut close_rx) = mpsc::channel::<Job>(1);
  let watcher = tokio::spawn(async move {
    assert!(close_rx.recv().await.is_none());
  });
  drop(close_tx);
  watcher.await.expect("watcher panicked");
  println!("[closed] 通道关闭后协作退出");

  // 独立接收端关闭例：发送失败，不等同于取消发送任务。
  let (rejected_tx, rejected_rx) = mpsc::channel::<Job>(1);
  drop(rejected_rx);
  let job = Job { title: "未投递".into(), kind: JobKind::Fast };
  assert!(rejected_tx.send(job).await.is_err());

  // 取消通道不投递 Job，发送端一直保留到取消结束。
  let (cancel_tx, mut cancel_rx) = mpsc::channel::<Job>(1);
  let (ready_tx, ready_rx) = oneshot::channel::<()>();
  let hung = tokio::spawn(async move {
    ready_tx.send(()).expect("ready receiver is alive");
    let _job = cancel_rx.recv().await;
  });
  ready_rx.await.expect("worker ready");
  hung.abort();
  let error = hung.await.expect_err("worker cancelled");
  assert!(error.is_cancelled());
  drop(cancel_tx);
  println!("[cancel] 任务被 abort 取消");
  println!("全部事件演示完毕: 完成={completed} 超时={timed_out}");
}

预期输出（正常运行；不承诺具体墙钟耗时）：

[worker] 完成: 索引重建
[worker] 完成: 缓存预热
[worker] 完成: 报表导出
[worker] 超时: 慢备份
[worker] 通道关闭，worker 退出
[closed] 通道关闭后协作退出
[cancel] 任务被 abort 取消
全部事件演示完毕: 完成=3 超时=1

timeout 的 Err 表示计时器条件导致包装 Future 未完成；它不是硬实时截止，若 Future 一次 poll 长时间不让出控制权，检查也可能延迟。这里三个 Fast 第一次 poll 就绪，Slow 永久等待，50ms 只决定等待上限的意图，不用于承诺机器上的精确耗时。参见 Tokio timeout。

协作关闭与 abort 是不同协议

项目中的 watcher 在通道关闭后正常结束，await 返回 Ok。接收端关闭使 send 返回 Err，由发送任务自行决定退出或处理消息，并不会取消任意任务。主处理循环也只是排空后协作退出。

取消例采用 ready 信号确认 worker 已启动；取消通道没有消息，发送端尚存，所以 recv 不会由于通道关闭而自然结束。abort 发出取消请求，随后 await 等待任务结束并断言 JoinError 的 is_cancelled。直到该结果确认后才销毁发送端，因此不存在“先关闭通道，再 abort，恰好自然结束”的竞态。

丢弃 Tokio JoinHandle 会 detach，任务继续运行。丢弃尚未完成的 Future 则释放其内部状态；析构不会撤销已经写出的文件、发送的请求或其他外部副作用。await 前后的状态应允许被中断，业务上的撤销要另有协议。abort 不是立即强杀，任务必须把控制权交回运行时，完成的任务也不能再被取消。

深入阅读：关闭、锁与测试的边界

无界队列在生产速度长期高于消费速度时可能持续占用内存。有界队列限制排队数量，但消息大小、上游缓冲和 worker 数量仍要单独约束。RwLock 也不保证比 Mutex 更快，需要基于实际读写负载测量。

标准库线程 handle 被丢弃也会脱离等待，panic 不能通过 join 返回给当前调用者。并发测试应覆盖空输入、发送端关闭、接收端关闭、worker panic 与取消顺序；重复运行只能增加观测机会，不能替代同步关系证明。第 21 章使用公共 API 的行为矩阵，第 19 章说明不要用 unsafe impl Send/Sync 绕过这些约束。