并发：让多个任务同时推进，又别把状态弄乱

你正在写一个批量处理程序：主线程读入一批文件，几个工作线程负责解析，最后再把结果汇总起来。单线程版本很好懂，可一旦你想“多开几个线程快一点”，问题马上就变了。

文件列表该借给线程，还是直接交给线程？工作线程崩了，主线程怎么知道？多个线程一起更新进度时，谁先改？任务生产得太快，内存会不会被待处理消息撑满？程序偶尔卡住，却没有任何报错，又该从哪里查？

并发难就难在这里：每一行局部代码都可能很正常，组合到一起却多出了时间顺序、资源归属和失败传播。Rust 不会假装这些问题很简单。它做的事情更像是安排了一位严格的代码评审员，在程序运行之前追问你：

• 这个线程可能活多久？
• 它拿走了什么，又借用了什么？
• 两个线程凭什么能同时接触这份数据？
• 一个线程失败后，其他线程准备继续、收尾，还是一起失败？

这些问题有时确实烦人，尤其是你第一次面对一长串 Send、Sync 和生命周期错误时。但编译器不是在跟你作对。它只是把原本可能在凌晨线上出现的问题，提前挪到了编辑器里。

先看两个编译不过的例子

先别急着背 Arc<Mutex<T>>。我们先亲手写出两段看起来很合理、实际却说不清安全边界的代码。等问题暴露出来，再看 Rust 为什么这样设计。

线程可能比借来的数据活得更久

下面的线程只想打印一个任务列表：

use std::thread;
 
fn main() {
    let jobs = vec!["解析配置", "生成报告", "清理缓存"];
 
    let handle = thread::spawn(|| {
        println!("{jobs:?}");
    });
 
    handle.join().unwrap();
}

你明明在后面调用了 join，人脑一看就知道主线程会等它，编译器却仍然拒绝。核心诊断大致是：

error[E0373]: closure may outlive the current function,
but it borrows `jobs`, which is owned by the current function

关键不在于你这次有没有记得 join，而在于 thread::spawn 创建的是一个可以脱离当前调用者继续存在的线程。它接收的闭包必须能独立活下去。普通借用的有效期只受词法作用域和生命周期约束保护，不能靠“我保证稍后会等它”这种运行时约定来延长。

最直接的修复是让闭包取得 jobs 的所有权：

use std::thread;
 
fn main() {
    let jobs = vec!["解析配置", "生成报告", "清理缓存"];
 
    let handle = thread::spawn(move || {
        println!("{jobs:?}");
    });
 
    handle.join().unwrap();
}

这里的 move 不是“让线程移动得更快”，也不是深拷贝。它让闭包按值捕获自己使用的外部变量。jobs 从此归闭包所有，主线程不能再使用它，而新线程也不再依赖主线程栈上的借用。

两个线程不能同时拿同一个可变借用

你可能接着想：既然作用域线程允许借用外部数据，那我让两个线程一起给计数器加一不就行了？

use std::thread;
 
fn main() {
    let mut total = 0;
 
    thread::scope(|scope| {
        scope.spawn(|| {
            total += 1;
        });
 
        scope.spawn(|| {
            total += 1;
        });
    });
 
    println!("{total}");
}

这次线程的生命周期没有问题，因为 scope 会等里面的线程结束。真正的问题是两个闭包都想在重叠的时间里持有 total 的可变借用。你会看到类似这样的核心诊断：

error[E0499]: cannot borrow `total` as mutable more than once at a time

如果 Rust 放行，两次“读取旧值、加一、写回”的步骤就可能交错。两个线程都读到 0，然后都写入 1，最终结果不是 2。更糟的是，对复杂数据结构进行未同步的并发读写会产生数据竞争，程序行为不再可靠。

编译器在这里没有替你选择解决方案，因为选择本身取决于业务：

• 如果每个线程可以处理互不重叠的数据，就把数据拆成互斥的片段分别借出。
• 如果线程只需要把结果交回来，就用通道传消息。
• 如果多个线程必须修改同一份状态，就用锁明确串行访问。
• 如果只是一个简单计数器或标志，并且你能说清内存顺序，再考虑原子类型。

这四条路线贯穿整章。你不需要一上来押注某个“最佳并发模型”，先看数据到底要怎么流动。

并发、并行、线程与异步不是一回事

日常交流里，这几个词经常混用。写程序时最好把它们分开，否则你很容易拿错工具。

并发关注的是多个任务能否在一段时间内共同推进。哪怕只有一个 CPU 核心，操作系统也可以在任务之间快速切换，让下载、解析和日志记录交错进行。

并行关注的是多个任务是否真的在同一时刻执行。要做到这一点，通常需要多个 CPU 核心，或者其他能并行计算的硬件。并发程序可能并行，也可能只是在单核上轮流运行。

操作系统线程是本章的主角。std::thread 创建的线程由操作系统调度，每个线程有自己的栈。线程适合运行会阻塞的工作，也适合把 CPU 密集计算分给多个核心，但创建和切换线程都有成本。为每个几毫秒的小任务单独创建一个线程，通常不是好主意。

异步任务是另一套调度模型。异步函数通常在执行器上协作运行，任务遇到可等待点时主动让出控制权。它很适合管理大量等待网络或磁盘事件的任务，但标准库本身不提供完整的异步执行器。异步也不自动等于多线程：一个执行器可以只用一个线程，多个异步任务也可能从未并行。

反过来也一样，多线程代码不等于异步代码。在线程里调用阻塞函数很常见；在异步执行器的工作线程里长时间阻塞，则可能拖住许多本来可以继续推进的任务。

并行不是免费的加速

把工作分到多个线程之前，先问三个很实际的问题：

• 工作能不能拆成相对独立的部分？
• 拆分、传递数据和最终合并的成本有多高？
• 多个线程会不会频繁争抢同一把锁？

如果任务本身很小，线程管理成本可能比计算还高。如果所有线程最后都排队修改一个全局 HashMap，你只是把单线程循环改造成了更复杂的单线程瓶颈。并发设计的重点不是“线程越多越快”，而是减少不必要的等待和共享。

线程数量也不该盲目照搬机器的逻辑核心数。CPU 密集任务、会阻塞的 I/O 任务、内存占用大的任务，需要的策略不同。标准库能给出可用并行度的估计，但它仍然只是设计输入，不是一个放到所有程序里都正确的常数。

先用最小可控的线程数跑通完整流程，再根据队列积压、处理时延和锁等待调整，通常比提前猜一个漂亮数字更可靠。

spawn、JoinHandle 与 move

thread::spawn 接收一个闭包，创建新线程，并立刻返回 JoinHandle<T>。闭包最后一个表达式的值就是线程的返回值，因此句柄里的 T 不一定是 ()。

use std::thread;
 
fn main() {
    let numbers = vec![3, 1, 4, 1, 5];
 
    let handle = thread::spawn(move || {
        numbers.into_iter().sum::<i32>()
    });
 
    let sum = handle.join().expect("计算线程发生 panic");
    println!("总和：{sum}");
}

总和：14

这里发生了几件事：

• move 把 numbers 的所有权交给闭包。
• 闭包返回 i32，所以 handle 的类型是 JoinHandle<i32>。
• join() 阻塞当前线程，直到目标线程结束。
• join() 消耗句柄，因为“等待这个线程结束”的权限是唯一的。
• 线程正常结束时，join() 返回 Ok(14)。

丢掉句柄不会自动等待

这是很容易被忽略的一点：普通 JoinHandle 被丢弃时，关联线程会被分离。它不会因为句柄离开作用域就自动 join。

分离后的线程可能继续运行，但你失去了等待它和取得返回值的办法。如果主线程随后结束，整个进程会终止，后台线程尚未完成的清理或写入也可能来不及执行。

所以，重要工作不要依赖“线程大概能跑完”。把句柄保存下来，在明确的收尾阶段逐个 join：

use std::thread;
 
fn main() {
    let mut handles = Vec::new();
 
    for worker_id in 0..4 {
        handles.push(thread::spawn(move || {
            worker_id * worker_id
        }));
    }
 
    let mut results = Vec::new();
 
    for handle in handles {
        results.push(handle.join().expect("工作线程发生 panic"));
    }
 
    println!("{results:?}");
}

[0, 1, 4, 9]

结果顺序之所以稳定，不是因为线程按编号完成，而是因为句柄按创建顺序放进了 Vec，主线程又按这个顺序逐个取返回值。线程真正完成的先后仍由调度器决定。

join 放在哪里会改变并发程度

下面两种写法看起来只差一行位置，行为却不同。

如果你每创建一个线程就立刻 join：

for id in 0..4 {
    let handle = std::thread::spawn(move || id * id);
    let result = handle.join().unwrap();
    println!("{result}");
}

循环必须等当前线程结束后才能创建下一个线程，工作基本被串行化了。

如果你先收集所有句柄，再统一等待，多个线程才有机会共同推进：

let handles: Vec<_> = (0..4)
    .map(|id| std::thread::spawn(move || id * id))
    .collect();
 
for handle in handles {
    println!("{}", handle.join().unwrap());
}

这不代表第二种一定更快，但它确实保留了并行执行的可能。

下面的时间线可以拖动查看 spawn、主线程继续执行和 join 等待之间的先后关系。重点观察 join 放置位置怎样改变两条执行路线的重叠区间。

move 移动的是捕获方式

move 闭包会按值捕获所使用的外部变量。对于 String、Vec<T> 这类非 Copy 类型，通常表现为所有权转移；对于 u32 这类 Copy 类型，则会复制值。

它不会自动把一份数据复制给每个线程。下面的循环仍然不成立：

use std::thread;
 
fn main() {
    let message = String::from("开始处理");
 
    for _ in 0..3 {
        thread::spawn(move || {
            println!("{message}");
        });
    }
}

第一次循环已经把 message 移进闭包，下一次循环没有东西可移。解决方法取决于含义：每个线程需要独立字符串，就在每轮先 clone；每个线程需要共享同一份不可变或受同步保护的数据，就考虑 Arc<T>；线程只需要收到一次通知，则通道往往更自然。

不要看到移动错误就条件反射地给所有类型套 Arc。编译器是在追问所有权设计，不是在暗示唯一答案。

子线程的 panic 怎样回到主线程

普通 Rust panic 通常只会让发生 panic 的线程开始展开。其他线程不会自动收到一份结构化错误。创建者需要通过 join() 检查：

use std::thread;
 
fn main() {
    let handle = thread::spawn(|| -> usize {
        panic!("输入数据损坏");
    });
 
    match handle.join() {
        Ok(value) => println!("结果：{value}"),
        Err(payload) => {
            if let Some(message) = payload.downcast_ref::<&str>() {
                println!("工作线程失败：{message}");
            } else if let Some(message) = payload.downcast_ref::<String>() {
                println!("工作线程失败：{message}");
            } else {
                println!("工作线程携带了无法识别的 panic 信息");
            }
        }
    }
}

默认 panic 钩子通常仍会先把 panic 信息打印到标准错误；join 的作用是让创建者知道线程失败，并取得 panic 载荷。你可以在这里决定记录失败、通知其他线程停止，或者让主线程继续 panic。

这里讨论的是会展开的 panic 策略。如果程序被配置为 panic 时直接中止，panic 会结束整个进程，也就没有机会通过 join 恢复载荷。

例如：

use std::io;
use std::thread;
 
fn load_count() -> io::Result<usize> {
    Ok(8)
}
 
fn main() -> io::Result<()> {
    let handle = thread::spawn(load_count);
 
    let count = handle
        .join()
        .expect("工作线程发生 panic")?;
 
    println!("读取到 {count} 条记录");
    Ok(())
}

这两层错误不要混成一个 unwrap()。线上程序通常需要分别制定策略。

线程也是一种需要预算的资源

thread::spawn 写起来像普通函数调用，很容易让人忘记它背后会创建操作系统线程。线程需要栈空间、调度时间和系统资源。任务数量一旦来自请求、文件数或用户输入，就不能简单地做到“一项任务一个线程”。

假设目录里平时只有几十个文件，这种写法看上去没有问题：

for path in paths {
    std::thread::spawn(move || process(path));
}

当目录突然变成几十万个文件时，程序会同时尝试创建几十万个线程。即使每个任务本身不重，线程栈和调度开销也足以把机器拖垮。更糟的是，这些线程的句柄没有保存，调用者既不知道它们何时结束，也不知道其中哪些 panic。

更可靠的设计会把“任务数量”和“执行线程数量”分开。任务可以很多，但只由一组数量受控的长期工作线程处理。任务通过有界队列进入，队列满时把压力反馈给生产者。这样，最大并发数和最大积压量都能在代码中看见。

可用并行度只是起点

标准库可以提供当前环境可用并行度的估计。它适合帮助设置 CPU 密集工作的初始线程数，但不是精确的性能答案。

如果工作主要等待磁盘或网络，线程数可能高于处理器并行度仍有收益；如果每个任务占用大量内存，哪怕线程数等于核心数也可能太多；如果多个线程争抢同一把锁，继续增加线程只会增加上下文切换。

所以线程数应该有默认值、允许配置，并在真实负载下观察。别把“机器有多少核”直接翻译成“程序永远创建多少线程”。

用 Builder 给线程取名并处理创建失败

thread::spawn 创建线程失败时会 panic。需要把系统资源不足作为可处理错误时，可以使用 thread::Builder，它的 spawn 返回 I/O 结果。Builder 也能设置线程名，线程转储和日志会更容易看懂。

use std::io;
use std::thread;
 
fn main() -> io::Result<()> {
    let handle = thread::Builder::new()
        .name(String::from("report-worker"))
        .spawn(|| {
            let current = thread::current();
            let name = current.name().unwrap_or("未命名线程");
            println!("正在运行：{name}");
            42
        })?;
 
    let value = handle
        .join()
        .expect("report-worker 发生 panic");
 
    println!("结果：{value}");
    Ok(())
}

正在运行：report-worker
结果：42

线程名是诊断信息，不是同步标识。不要根据名字决定锁权限或消息路由。Builder 还允许设置栈大小，但这属于需要测量的高级调整：栈太小可能溢出，盲目调大又会增加资源压力。

JoinHandle 不是取消按钮

句柄可以等待线程、读取返回值和查看线程信息，却不能安全地强行终止目标线程。原因很现实：线程可能正拿着锁、写到一半，或者暂时破坏了某个只在函数结束时恢复的不变量。

需要取消时，应让线程协作退出。常见方式是：

• 关闭任务发送端，让工作线程在队列耗尽后结束；
• 发送明确的停止命令；
• 设置原子取消标志，让长任务在安全检查点退出；
• 让任务函数返回错误，由上层决定是否继续派发。

取消请求发出后，创建者仍要 join。请求退出只表达意图，join 才确认线程真的结束并完成清理。若你只设置标志就丢掉句柄，线程仍可能因为没走到检查点而长期存活。

作用域线程：让线程安全地借用局部数据

普通 thread::spawn 的线程可能比创建它的函数活得更久，所以闭包通常需要满足 'static 边界。这里的 'static 不等于线程会永远运行，它表示闭包不能携带寿命不足的借用。

有些任务明明可以在一个明确的代码块内完成，让每个线程都取得数据所有权反而很别扭。比如你想把一个切片分成两半，各自原地处理。thread::scope 正是为这种场景准备的。

use std::thread;
 
fn main() {
    let mut numbers = [1, 2, 3, 4, 5, 6];
 
    thread::scope(|scope| {
        let (left, right) = numbers.split_at_mut(3);
 
        scope.spawn(move || {
            for value in left {
                *value *= 10;
            }
        });
 
        scope.spawn(move || {
            for value in right {
                *value += 100;
            }
        });
    });
 
    assert_eq!(numbers, [10, 20, 30, 104, 105, 106]);
}

这段代码没有锁，却能并发修改一个数组。安全的理由不是“两个线程恰好不会撞上”，而是 split_at_mut 已经在类型层面把数组拆成两个不重叠的可变切片。每个线程只拿到自己那半份。

作用域结束前，尚未手动等待的作用域线程都会被等待。因此当 thread::scope 返回时，numbers 不再被线程借用，主线程可以继续访问它。

作用域不是绕过借用检查

作用域线程放宽的是生命周期限制，不是别名规则。前面两个线程同时修改 total 的例子仍然会失败，因为两个可变借用发生了重叠。

换句话说：

• 普通线程要求数据能够独立活得足够久，常见做法是移动所有权。
• 作用域线程允许借用外部数据，因为它保证所有线程在作用域退出前结束。
• 无论哪一种线程，都不能凭空获得未同步的共享可变访问。

作用域里的 panic 仍需设计

scope.spawn 返回 ScopedJoinHandle<T>，你可以手动 join 并处理 panic。如果你丢掉作用域句柄，作用域结束时会隐式等待它；若这类被隐式等待的线程发生 panic，thread::scope 会在等待其他线程结束后继续 panic。

这比普通句柄“丢掉就分离”更适合结构化并发：子线程不会悄悄越过父作用域。但它仍然不替你决定失败策略。只要一个子任务失败，是取消全批任务、保留已完成结果，还是重试失败部分，依然要由业务代码说明。

什么时候优先使用作用域线程

作用域线程特别适合这些情况：

• 将一个大切片安全地拆成互不重叠的部分。
• 并行读取当前函数里的只读配置，不想为了生命周期而复制。
• 明确要求所有子任务在函数返回前完成。
• 希望借用检查器帮助证明子任务没有逃逸。

如果线程需要长期存活，组成常驻工作池，或者跨越当前请求的生命周期，普通线程配合拥有的数据和通道通常更合适。

消息传递：先把数据归属说清楚

假设多个工作线程各自计算结果，主线程只负责汇总。让所有线程共同修改一个结果容器当然可以，但那意味着每次写入都要竞争锁。另一种思路是：工作线程完成后把结果发送给汇总者，发送之后就不再碰它。

标准库的 std::sync::mpsc 提供多生产者、单消费者通道：

• Sender<T> 或 SyncSender<T> 可以克隆，因此多个线程能发送。
• Receiver<T> 只有一个，负责接收。
• send(value) 按值接收消息，非 Copy 值发送成功后，发送方不能继续使用。

这套模型的直觉很像快递：多个发件人可以往同一个收件地址寄包裹，但包裹交出去后不再归发件人保管。

多个生产者汇总结果

use std::sync::mpsc;
use std::thread;
 
#[derive(Debug)]
struct Report {
    worker_id: usize,
    score: usize,
}
 
fn main() {
    let (tx, rx) = mpsc::channel();
    let mut handles = Vec::new();
 
    for worker_id in 0..4 {
        let worker_tx = tx.clone();
 
        handles.push(thread::spawn(move || {
            let report = Report {
                worker_id,
                score: worker_id * worker_id,
            };
 
            worker_tx.send(report).expect("汇总线程已退出");
        }));
    }
 
    drop(tx);
 
    let mut reports: Vec<_> = rx.into_iter().collect();
    reports.sort_by_key(|report| report.worker_id);
 
    for handle in handles {
        handle.join().expect("工作线程发生 panic");
    }
 
    for report in reports {
        println!("工作线程 {}：{}", report.worker_id, report.score);
    }
}

工作线程 0：0
工作线程 1：1
工作线程 2：4
工作线程 3：9

工作线程完成的顺序不确定，所以接收顺序也不能用来代表 worker_id 顺序。示例先排序再输出，才得到稳定结果。

同一个发送者按顺序发送的消息会按顺序进入接收端。多个发送者同时发送时，谁先成功没有固定保证。业务如果依赖全局顺序，应在消息中携带序号、时间窗口或其他可验证的排序信息，而不是猜调度顺序。

为什么必须 drop(tx)

rx.into_iter() 会持续等待新消息，直到所有发送端都被丢弃，并且缓冲区中已有消息已经取完。

循环里创建的 worker_tx 会在线程结束时自动丢弃，但主线程手里还保留着最初的 tx。如果不显式 drop(tx)，接收端会认为未来仍可能有消息，迭代就一直等下去。

这是一种很实用的终止协议：发送端全部消失，就表示数据流结束。它比额外发送一个 "STOP" 字符串更不容易撞上业务数据，但前提是你确实管理好了每一个发送端克隆。某个克隆被长期保存在结构体里，接收循环就不会结束。

断开连接也是信息

通道两端的操作返回 Result，断开时不会假装成功：

• 所有发送端都被丢弃后，recv() 在取完缓冲消息后返回 RecvError。
• 接收端被丢弃后，后续 send(value) 返回 SendError<T>，并把未发送的值带回来。
• try_recv() 不阻塞，它会区分“现在暂时没有消息”和“发送端已经全部断开”。
• recv_timeout() 最多等待指定时长，它会区分超时和断开。

生产者发现接收端消失时，应该决定任务是否还有继续计算的意义。如果汇总者已经退出，继续消耗 CPU 生成没人接收的结果往往只是浪费。相反，消费者发现发送端全部消失时，要区分“正常完成”和“生产者意外退出”。只看通道断开本身，无法知道是哪一种；你可以同时检查线程句柄，或者在消息协议里发送明确的完成状态。

channel 没有背压

mpsc::channel() 创建的是概念上的无界缓冲通道。发送不会因为缓冲区满而阻塞，因为它没有你指定的固定容量。

“无界”不等于机器内存无限。假设生产者每秒生成十万条消息，消费者每秒只能处理一万条，积压会不断增长，最终表现为内存上升、延迟变长，甚至进程被系统终止。

这种通道适合消息量明确很小、生产速度天然受限，或者发送方不能被阻塞的场景。只要生产速度可能持续高于消费速度，你就需要考虑背压。

sync_channel：让生产速度受消费能力约束

mpsc::sync_channel(capacity) 创建有界同步通道。缓冲区满时，send 会阻塞，直到消费者腾出位置；这就是最直接的背压。

use std::sync::mpsc;
use std::thread;
use std::time::Duration;
 
fn main() {
    let (tx, rx) = mpsc::sync_channel(2);
 
    let producer = thread::spawn(move || {
        for job_id in 0..5 {
            tx.send(job_id).expect("消费者已退出");
        }
    });
 
    for job_id in rx {
        println!("处理任务 {job_id}");
        thread::sleep(Duration::from_millis(20));
    }
 
    producer.join().expect("生产线程发生 panic");
}

容量为 2 表示最多允许两条待接收消息留在缓冲区里。消费者处理得慢时，生产者会在 send 处等待，不会无限堆积。

这里的 sleep 只是模拟慢处理，不承担同步正确性。删掉它，程序仍然正确，只是背压不一定明显。

你可以在下面调整通道容量与生产、消费速度，观察队列积压和发送阻塞如何变化。把容量降到零时，再留意发送与接收必须直接会合的效果。

容量为零是一手交一手

sync_channel(0) 没有缓冲位。每次发送都必须等某个接收操作与它配对，数据才算交接完成。这种通道常被称为会合通道。

它能提供很强的节奏约束，但也更容易因为双方都在等而卡住。比如同一线程先调用 send，准备稍后再 recv，零容量通道会让它永远走不到接收那一行。发送和接收必须由可以分别推进的执行路线完成。

不想阻塞时使用 try_send

SyncSender::try_send 会立刻返回。失败分两种：

use std::sync::mpsc::{self, TrySendError};
 
fn main() {
    let (tx, _rx) = mpsc::sync_channel(1);
 
    tx.send("第一条").unwrap();
 
    match tx.try_send("第二条") {
        Ok(()) => println!("已发送"),
        Err(TrySendError::Full(value)) => {
            println!("队列已满，稍后处理：{value}");
        }
        Err(TrySendError::Disconnected(value)) => {
            println!("接收端已退出，未发送：{value}");
        }
    }
}

队列已满，稍后处理：第二条

遇到 Full 以后怎么做，不能靠通道替你决定。你可以等待重试、丢弃低优先级消息、覆盖旧状态，或者把压力反馈给上游。每一种选择都改变业务语义。

两类通道的选择

可以先用这个朴素判断：

• 消息很少、峰值可控，发送方不能停：考虑 channel。
• 消费能力有限，必须限制积压：考虑 sync_channel。
• 希望发送和接收严格会合：考虑容量为零的 sync_channel。
• 需要多个消费者同时从一个队列抢任务：标准库 mpsc 的接收端不是多消费者设计，需要重新设计分发方式，而不是简单克隆 Receiver。

有些代码会把 Receiver 放进 Arc<Mutex<_>>，让多个工作线程轮流锁住后 recv。它可以工作，但锁住接收端等待消息时，其他工作线程也无法接收，行为和吞吐未必符合你想象。更清楚的做法往往是让一个分发线程持有接收端，再把任务明确路由给工作线程，或者选择与需求匹配的队列实现。

Arc<Mutex<T>>：共享所有权加独占访问

有些状态天生就是共享的。比如多个工作线程都要更新同一组指标，最终又要由主线程读取。此时 Arc<Mutex<T>> 是标准库里最常见的组合。

可以把两层职责分开理解：

• Arc<T> 解决“谁拥有这份数据”。克隆 Arc 只会增加共享所有权，不会复制内部的 T。
• Mutex<T> 解决“谁现在可以访问可变数据”。拿到锁的线程获得 MutexGuard<T>，其他线程必须等待。

Arc 里的原子引用计数只保护引用计数本身。它不会自动让内部数据可以并发修改。因此 Arc<Vec<_>> 允许多个线程共享只读访问，却不能让多个线程直接 push。要修改，仍需 Mutex、RwLock、原子类型，或者把修改转为消息。

从移动错误走到共享所有权

下面是一个完整计数例子：

use std::sync::{Arc, Mutex};
use std::thread;
 
fn main() {
    let buckets = Arc::new(Mutex::new(vec![0_usize; 4]));
    let mut handles = Vec::new();
 
    for index in 0..8 {
        let buckets = Arc::clone(&buckets);
 
        handles.push(thread::spawn(move || {
            let bucket = index % 4;
            let mut guard = buckets.lock().expect("计数锁已中毒");
            guard[bucket] += 1;
        }));
    }
 
    for handle in handles {
        handle.join().expect("工作线程发生 panic");
    }
 
    let snapshot = buckets
        .lock()
        .expect("计数锁已中毒")
        .clone();
 
    assert_eq!(snapshot, vec![2, 2, 2, 2]);
}

每轮循环都克隆一个 Arc，再用 move 把这个克隆交给线程。所有克隆指向同一个 Mutex<Vec<usize>>。

lock() 返回 Result<MutexGuard<_>, PoisonError<_>>。获得 guard 后，DerefMut 让我们能像操作 &mut Vec<_> 一样修改数据。guard 离开作用域就自动释放锁，不需要手写 unlock。

下面的交互工坊把 Arc 克隆、锁竞争与 guard 释放拆开呈现。切换不同线程的执行顺序，看看共享所有权和独占访问为什么必须分别负责。

临界区越短越好，但别为了短而破坏一致性

锁被持有的代码范围叫临界区。常见建议是缩短临界区，因为锁持有得越久，其他线程等得越久。

比如你要先做耗时计算，再更新汇总值，应该把计算放到锁外：

let partial = expensive_calculation();
 
{
    let mut total = total.lock().unwrap();
    *total += partial;
}

不要在持锁期间做这些事，除非你非常确定：

• 阻塞接收通道消息；
• 等待另一个线程 join；
• 长时间睡眠；
• 调用可能回调到当前模块的外部代码；
• 执行与被保护状态无关的耗时计算。

不过，“缩短锁”不能以破坏业务不变量为代价。如果余额扣减和流水写入必须一起成功，就不能为了减少几行临界区把它们拆成两个可被其他线程插入的阶段。正确性先于竞争优化。

用显式代码块尽早释放 guard

有时 guard 的生命周期比你直觉中长。最清楚的做法是用代码块明确边界：

let current = {
    let guard = shared_state.lock().unwrap();
    guard.current_value
};
 
send_result(current);

进入 send_result 前锁已经释放。不要只依赖某个临时值“应该已经被编译器丢掉”；并发代码里，显式边界更容易评审。

Mutex 不保证业务公平

互斥锁保证同一时刻最多一个线程持有 guard，但它不承诺等待线程严格按先来后到的顺序获得锁。某个线程能不能很快拿到锁，受操作系统调度和实现策略影响。

如果业务要求明确排队顺序，应该把请求放进带顺序语义的队列，由一个拥有状态的线程处理。不要把锁的唤醒顺序当成业务协议。

RwLock<T>：允许多个读者，但写者仍然独占

Mutex<T> 不区分读和写。只要一个线程拿到锁，其他线程都要等。对于读操作很多、写操作较少，而且每次读取不算太短的场景，RwLock<T> 可能减少读者之间的竞争。

它提供两类 guard：

• read() 获得共享读 guard，可以同时存在多个。
• write() 获得独占写 guard，存在写 guard 时不能有其他读写 guard。

use std::sync::{Arc, RwLock};
use std::thread;
 
fn main() {
    let config = Arc::new(RwLock::new(String::from("v1")));
    let mut readers = Vec::new();
 
    for _ in 0..3 {
        let config = Arc::clone(&config);
 
        readers.push(thread::spawn(move || {
            let snapshot = config
                .read()
                .expect("配置锁已中毒")
                .clone();
 
            snapshot.len()
        }));
    }
 
    for reader in readers {
        assert_eq!(reader.join().unwrap(), 2);
    }
 
    {
        let mut config = config.write().expect("配置锁已中毒");
        *config = String::from("v2");
    }
}

示例在读锁内只克隆快照，随即释放锁。线程后续若要做耗时处理，可以基于自己的快照继续，不必长期挡住写者。

RwLock 不一定比 Mutex 快

如果每次读取只有一次整数访问，维护读者状态的成本可能抵消并发读取的收益。如果写入频繁，读者和写者会不断互相阻塞，RwLock 甚至可能更慢。

更关键的是，标准库的 RwLock 不保证固定的读写优先策略。等待中的写者是否会阻止后来读者继续获得读锁，取决于底层系统实现。你不能依赖“写者一定优先”，也不能依赖“读者一定不会饿死”。

选择 RwLock 前，先确认这些条件：

• 读操作明显多于写操作；
• 读临界区足够大，允许多个读者确实有收益；
• 业务不依赖特定的锁获取顺序；
• 你愿意用真实负载测量，而不是根据名字猜性能。

不要试图把读锁直接升级成写锁

一个常见陷阱是先持有读锁检查状态，发现需要修改后，又在读 guard 尚未释放时调用 write()。写锁必须等所有读锁离开，其中也包括你自己手上的那个，于是线程可能把自己等住。

安全的基本写法是先读取并复制决定所需的信息，释放读 guard，再申请写锁，然后重新检查条件：

let needs_update = {
    let state = shared.read().unwrap();
    state.version < expected_version
};
 
if needs_update {
    let mut state = shared.write().unwrap();
 
    if state.version < expected_version {
        state.version = expected_version;
    }
}

第二次检查不能省。释放读锁到获得写锁之间，其他线程可能已经修改了状态。

死锁：程序没崩，但所有人都在等

Rust 的类型系统能阻止数据竞争，却不能自动阻止死锁。死锁是逻辑层面的循环等待：线程甲拿着资源 A 等资源 B，线程乙拿着资源 B 等资源 A，谁都不肯先放手。

下面的转账函数本身能编译：

use std::sync::Mutex;
use std::thread;
 
fn transfer(from: &Mutex<i64>, to: &Mutex<i64>, amount: i64) {
    let mut from_balance = from.lock().unwrap();
    let mut to_balance = to.lock().unwrap();
 
    *from_balance -= amount;
    *to_balance += amount;
}
 
fn main() {
    let account_a = Mutex::new(100);
    let account_b = Mutex::new(100);
 
    thread::scope(|scope| {
        scope.spawn(|| transfer(&account_a, &account_b, 10));
        scope.spawn(|| transfer(&account_b, &account_a, 20));
    });
}

它不保证每次都死锁。可一旦第一个线程先拿到 A、第二个线程先拿到 B，双方再去拿对方手里的锁，就会永远等待。最让人难受的是：程序没有 panic，也没有编译错误，只是偶尔停在那里。

固定锁顺序

最有效的基本规则是：所有代码都按同一个全局顺序获取多把锁。比如每个账户有稳定且唯一的 ID，总是先锁 ID 小的账户，再锁 ID 大的账户。业务上的“转出”和“转入”方向不应该决定加锁顺序。

这条规则要写进封装边界，而不是只留在团队记忆里。最好让“同时拿两把账户锁”的动作集中在一个函数中，其他代码没有机会发明自己的顺序。

下面的检测器允许你组合线程、锁与等待边，观察何时出现闭环。尝试把两条路径改成相同加锁顺序，比较等待图怎样变化。

避免持锁等待未知动作

通道接收、线程等待和外部回调都可能无限期停住。持锁执行这些动作，会把当前等待扩散给所有需要该锁的线程。

一个很实用的排查问题是：我拿着这把锁时，下一行代码是否可能等待另一条执行路线？

如果答案是“可能”，继续追问那条执行路线是否也可能需要当前锁。很多死锁就是这样被找到的。

try_lock 能探测竞争，但不是万能修复

try_lock() 不会阻塞。锁正被持有时，它会返回 WouldBlock。你可以释放已经拿到的其他锁，稍后重试，从而避免直接形成循环等待。

但重试策略可能带来活锁：两个线程反复同时拿起、同时失败、同时放下，程序一直忙却没有进展。退避和随机等待能缓解部分情况，却不能替代清楚的资源顺序。

不要用 sleep 修死锁

把某个线程睡几毫秒，只是改变了这一次调度概率。机器更快、负载更高、日志多一行，原来的交错就可能回来。sleep 可以模拟耗时，不能证明同步关系。

真正的同步要靠所有权、通道、锁、条件变量、原子操作或线程等待这些有明确语义的工具。

锁中毒：有人在修改中途 panic 了

假设线程拿到 Mutex 后，先改了数据的一半，随后 panic。锁的 guard 在展开过程中会被丢弃，因此互斥锁本身可以再次获得。但内部数据是否仍满足业务不变量，已经不确定。

标准库的 Mutex 会记录这种情况，后续 lock() 返回 PoisonError。这就是锁中毒。它不是说锁坏了，而是在提醒你：上一个独占访问者没有正常走完临界区。

很多教学示例直接写：

let guard = state.lock().unwrap();

这样做会在锁中毒时继续 panic，把失败传播出去。对于“任何部分更新都不可接受”的程序，这可能正是合理策略。可是它不是唯一策略。

恢复之前先证明不变量

PoisonError::into_inner() 可以取回 guard，说明中毒是建议性的保护，不是永久封锁。问题是，你不能因为 API 允许恢复，就假设数据正确。

let mut guard = match state.lock() {
    Ok(guard) => guard,
    Err(poisoned) => {
        let mut guard = poisoned.into_inner();
 
        // 只有在你能证明这样会恢复一致性时，才能继续。
        guard.rebuild_from_known_good_state();
        guard
    }
};
 
guard.accept_next_update();

这里的两个方法代表你的业务修复逻辑。真实程序可能从持久化快照重建、回滚未提交操作，或者直接拒绝服务。若你无法证明数据恢复一致，就不要继续把它交给其他线程。

锁中毒也不是内存安全的唯一防线，更不能保证捕获每一种异常路径。它只是标准锁提供的一条故障信号。

RwLock 的中毒范围不同

标准库 RwLock 只有在线程持有写锁时发生 panic，才会中毒。读者 panic 不会把锁标记为中毒，因为读 guard 本来不应该修改受保护数据。

这再次说明 unwrap() 背后藏着策略。原型程序可以选择遇到中毒就整体失败；长期运行的服务需要明确记录、隔离或恢复流程。

panic、通道断开和中毒要连起来看

一次工作线程 panic 可能同时产生三种外部信号：

• 它的 JoinHandle::join() 返回 Err；
• 它持有的发送端被丢弃，接收者最终看到通道断开；
• 它若在持有独占锁时 panic，后续加锁者看到中毒。

这三种信号描述的是同一次失败的不同侧面。一个可靠的协调者应该决定哪个是主要故障信号，避免把同一个故障记录成三次互不相关的事故。

Send 与 Sync：类型能不能跨过线程边界

你会在并发错误里频繁看到 Send 和 Sync。它们没有要调用的方法，是用于表达线程安全边界的标记 trait。

先记住两个直觉：

• T: Send 表示把一个 T 的所有权交给另一线程是安全的。
• T: Sync 表示多个线程共享 &T 是安全的。

更精确地说，如果 &T 可以安全地发送到另一线程，那么 T 就是 Sync。

为什么 Rc<T> 不能直接跨线程

Rc<T> 的引用计数不是原子更新。两个线程同时克隆或丢弃相连的 Rc，可能同时修改引用计数。因此 Rc<T> 既不适合被发送到另一线程，也不适合由多个线程共享。

Arc<T> 使用适合多线程的原子引用计数，所以它能承担共享所有权。但别漏掉后半句：Arc<T> 能否 Send 或 Sync，还取决于内部的 T。

例如，RefCell<T> 的借用规则在运行时检查，这套检查不是线程同步机制。给它套上 Arc 并不会把它变成多线程锁：

use std::cell::RefCell;
use std::sync::Arc;
use std::thread;
 
fn main() {
    let values = Arc::new(RefCell::new(vec![1, 2, 3]));
    let worker_values = Arc::clone(&values);
 
    thread::spawn(move || {
        worker_values.borrow_mut().push(4);
    });
}

这段代码不会通过编译。问题不是 Arc，而是内部 RefCell<Vec<_>> 不能被多个线程安全共享。多线程共享可变状态应使用线程安全的同步原语，例如 Arc<Mutex<T>>。

自动推导让边界沿类型结构传播

绝大多数由 Send 字段组成的普通结构体会自动成为 Send；由 Sync 字段组成的结构体会自动成为 Sync。因此你通常不需要，也不应该手动实现它们。

可以用两个只在编译期检查的辅助函数观察类型：

use std::sync::{Arc, Mutex};
 
fn assert_send<T: Send>() {}
fn assert_sync<T: Sync>() {}
 
fn main() {
    assert_send::<String>();
    assert_sync::<String>();
    assert_send::<Arc<Mutex<Vec<i32>>>>();
    assert_sync::<Arc<Mutex<Vec<i32>>>>();
}

如果某个类型不满足边界，函数不需要真正运行，编译器就会指出是哪个组成部分阻止了推导。

Send 和 Sync 不表示业务逻辑正确

一个类型满足 Send + Sync，只表示按其安全 API 跨线程使用不会造成 Rust 所定义的数据竞争或内存不安全。它不保证：

• 不会死锁；
• 消息不会重复处理；
• 计数一定符合业务规则；
• 多步更新具有事务性；
• 线程不会饥饿；
• 程序一定比单线程快。

这层边界很重要，但别把它理解成并发正确性的毕业证。

原子类型：适合小状态，不适合含糊的大协议

锁通过 guard 保护一段临界区，原子类型则对单个整数、布尔值或指针提供不可分割的操作。常见类型包括 AtomicBool、AtomicUsize 和 AtomicIsize。

一个简单计数器可以这样写：

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
 
fn main() {
    let completed = Arc::new(AtomicUsize::new(0));
    let mut handles = Vec::new();
 
    for _ in 0..8 {
        let completed = Arc::clone(&completed);
 
        handles.push(thread::spawn(move || {
            completed.fetch_add(1, Ordering::Relaxed);
        }));
    }
 
    for handle in handles {
        handle.join().unwrap();
    }
 
    assert_eq!(completed.load(Ordering::Relaxed), 8);
}

fetch_add 会原子地完成“读取旧值并增加”的操作，不会出现两个线程覆盖彼此更新。这里使用 Relaxed，因为计数器只表达完成次数，不负责宣布其他非原子数据已经准备好；主线程又在读取前等待了所有工作线程结束。

原子操作不等于一组操作组成事务

假设你有两个原子字段：

use std::sync::atomic::AtomicUsize;
 
struct Inventory {
    available: AtomicUsize,
    reserved: AtomicUsize,
}

分别修改它们是原子的，不代表“库存减少并且预留增加”作为整体不可分割。其他线程可能在两个操作之间观察到中间状态。

一旦业务不变量跨越多个字段，Mutex 往往更容易证明正确。原子类型的代码可能没有锁，却把复杂性转移到了内存顺序、重试循环和状态机上。行数少不等于风险小。

Ordering 到底在约束什么

原子操作需要一个内存顺序参数。它不只是“性能档位”，而是在说明不同线程对内存操作的可见顺序。

入门阶段可以先建立下面的边界：

• Relaxed 只保证当前原子操作不可分割，不用它发布其他普通数据。
• Release 常用于发布端，表示它之前的操作不能跑到这次发布之后。
• Acquire 常用于接收端；当它观察到相匹配的发布后，发布前的写入对接收端可见。
• AcqRel 用在同时读写的原子操作上，结合获取和释放语义。
• SeqCst 在相应获取、释放效果之外，还让所有线程对顺序一致操作形成统一的全局顺序。

这几句话只是地图，不足以让你安全地发明无锁数据结构。真正使用 Acquire/Release 时，你必须说清哪个原子写与哪个原子读配对，以及哪批普通数据借此发布。

SeqCst 更容易推理，但不是万能修复

当你还无法证明更弱顺序足够时，SeqCst 往往是更稳妥的起点。等性能测量表明确有必要，再在完整论证和测试下减弱顺序。

不过，SeqCst 只约束相关原子操作。它不会：

• 把普通的未同步共享写入变合法；
• 让两个原子字段自动成为事务；
• 防止死锁或活锁；
• 修复错误的状态机；
• 替代通道的容量与终止协议。

如果你发现自己靠不断把顺序改强、改弱来碰运气，先退回 Mutex 或通道。能清楚说明正确性，比“看起来无锁”更重要。

自旋也会浪费 CPU

不断循环读取 AtomicBool 等待状态变化，写起来很短，却可能占满一个 CPU 核心。短暂且经过测量的底层等待才可能适合自旋。普通应用若要等待较长时间，阻塞通道、条件变量、线程停放或其他有休眠语义的机制通常更合适。

原子类型最可靠的入门用途是独立计数、简单标志和已被成熟算法封装的底层组件。跨字段协议、无锁队列和内存回收已经进入高级并发领域，不要把它们当成“少写一把锁”。

消息传递还是共享状态

这两种模型没有谁天然更高级。它们只是把复杂性放在不同位置。

消息传递的手感

消息传递让所有权随消息流动，组件之间边界清楚。它适合：

• 任务分发和结果汇总；
• 单一所有者维护状态，其他线程提交命令；
• 流水线式处理；
• 希望通过断开通道表达结束。

它的代价包括数据移动或复制、队列内存、额外延迟，以及需要设计容量和关闭协议。通道也可能死锁：发送者等容量，接收者又在等发送者持有的其他资源，就形成了另一种循环等待。

共享状态的手感

共享状态适合：

• 多个线程频繁读取同一份配置；
• 很小的共享指标；
• 修改动作天然围绕一个共同数据结构；
• 复制或转移数据成本太高。

它的代价是锁竞争、死锁风险、中毒处理，以及调用链深处不易看见的锁持有范围。共享状态一多，局部函数很难知道自己是否处在某个锁的临界区里。

一个实用的判断流程

很多成熟设计会混用两者。例如，工作线程通过有界通道接收任务，结果通过另一个通道交给汇总线程；全局取消标志使用原子布尔值；少量配置放在 RwLock 中。混用没有问题，只要每个同步点都有单一、清楚的职责。

一套能落地的排错顺序

并发 bug 最大的麻烦是它不一定每次出现。日志一加，调度变了，问题又消失。你需要比“多跑几次看看”更稳定的检查顺序。

先检查生命周期和所有权

问清楚每个线程持有什么：

• 是独立值、共享引用，还是 Arc 克隆？
• 普通线程是否错误地借用了局部变量？
• 某个 move 是否把后续仍需使用的值拿走了？
• 作用域线程借用的区域是否真正互不重叠？

编译器错误通常已经沿类型链指出问题。先读最底层的“哪个类型没有实现 Send 或 Sync”，不要只盯着最上面的泛型错误。

再列出所有可能阻塞的位置

把这些调用标出来：

• join
• Mutex::lock
• RwLock::read 和 write
• Receiver::recv
• SyncSender::send
• 任何文件、网络或外部进程操作

然后为每个位置写一句“它在等谁”。如果等待关系绕一圈回到自己，就找到了死锁候选。

检查关闭协议

通道循环不结束时，别急着怀疑 recv。先确认所有发送端克隆是否都被丢弃。工作线程不退出时，确认它是否仍在等一个永远不会来的任务。主线程 join 卡住时，检查它是否还持有子线程退出所需的锁或发送端。

把锁范围变得可见

给 guard 使用独立代码块。避免把 lock().unwrap() 直接嵌进很长的表达式。复杂函数可以先复制小快照，再离开锁范围。

如果函数要求调用者已经持有某把锁，尽量通过参数类型或命名表达出来。靠注释说“调用前必须加锁”，很容易在重构后失效。

不要验证固定输出顺序

除非你明确建立了顺序同步，否则线程日志顺序就是不稳定的。测试应该验证集合内容、最终状态或协议事件，而不是“线程 A 一定先打印”。

需要稳定比较结果时，可以像前面的通道例子那样，给消息携带 ID，收集后排序。不要用 sleep 把一次偶然顺序伪装成契约。

给失败安排收尾路线

每个线程入口最好回答这些问题：

• 常规错误如何返回？
• panic 由谁 join 并记录？
• 某个工作线程退出后，其他线程怎样得知？
• 通道发送失败时，未发送值如何处理？
• 锁中毒后是停止、重建，还是确认一致后继续？

并发程序的成功路线可能很短，真正决定可靠性的往往是失败路线。

从需求走到代码：一个有界批处理器

前面的例子一次只解释一个工具。真实程序不会这么体贴，它通常把问题一起扔给你。

假设我们要做一个批处理器：

• 主线程有一批待计算任务；
• 三个工作线程共同处理；
• 每个工作线程的待办不能无限增长；
• 结果也要限制积压；
• 全部任务完成后，所有线程都应自然退出；
• 任意线程 panic，协调者必须知道。

标准库 mpsc 的接收端不能克隆，所以我们不能创建一个任务接收端，再让三个工作线程直接抢。这里采用一个很直白的设计：每个工作线程拥有自己的有界任务通道，分发线程按轮转顺序投递；所有工作线程再把结果发到同一个有界结果通道。

use std::sync::mpsc;
use std::thread;
 
#[derive(Debug)]
struct Job {
    id: usize,
    input: u64,
}
 
#[derive(Debug)]
struct Completed {
    job_id: usize,
    worker_id: usize,
    output: u64,
}
 
fn main() {
    let jobs: Vec<_> = (0..12)
        .map(|id| Job {
            id,
            input: id as u64,
        })
        .collect();
 
    let worker_count = 3;
    let (result_tx, result_rx) = mpsc::sync_channel::<Completed>(4);
    let mut job_txs = Vec::new();
    let mut worker_handles = Vec::new();
 
    for worker_id in 0..worker_count {
        let (job_tx, job_rx) = mpsc::sync_channel::<Job>(2);
        let result_tx = result_tx.clone();
 
        job_txs.push(job_tx);
 
        worker_handles.push(thread::spawn(move || {
            for job in job_rx {
                let completed = Completed {
                    job_id: job.id,
                    worker_id,
                    output: job.input * job.input,
                };
 
                if result_tx.send(completed).is_err() {
                    break;
                }
            }
        }));
    }
 
    drop(result_tx);
 
    let dispatcher = thread::spawn(move || -> Result<(), Job> {
        for (index, job) in jobs.into_iter().enumerate() {
            let worker_index = index % job_txs.len();
 
            if let Err(error) = job_txs[worker_index].send(job) {
                return Err(error.0);
            }
        }
 
        Ok(())
    });
 
    let mut completed: Vec<_> = result_rx.into_iter().collect();
 
    let dispatch_result = dispatcher
        .join()
        .expect("分发线程发生 panic");
 
    for handle in worker_handles {
        handle.join().expect("工作线程发生 panic");
    }
 
    if let Err(job) = dispatch_result {
        panic!("任务 {} 未能送达工作线程", job.id);
    }
 
    completed.sort_by_key(|item| item.job_id);
 
    for item in completed {
        println!(
            "任务 {} 由工作线程 {} 完成，结果 {}",
            item.job_id,
            item.worker_id,
            item.output
        );
    }
}

这段代码比“创建三个线程然后加锁”长一些，但每一段等待关系都能讲清楚。

为什么分发也放进单独线程

任务通道容量为 2，结果通道容量为 4。如果主线程先把所有任务发送完，之后才接收结果，可能出现这样的等待环：

1. 主线程在某个已满的任务通道上等待；
2. 对应工作线程处理完任务，准备发送结果；
3. 结果通道已经满了，工作线程也开始等待；
4. 唯一会接收结果的主线程仍卡在发送任务。

两边都在正确使用有界通道，组合起来却仍然能死锁。背压只定义一次发送怎样等待，不会替你检查整个等待图。

示例让分发线程专门发送任务，主线程从一开始就接收结果。生产路径和消费路径可以同时推进，等待环被拆开。你也可以在一个事件循环里交替尝试发送和接收，但代码会更复杂。

为什么要丢掉最初的结果发送端

每个工作线程都拿到一个 result_tx 克隆。主线程创建通道时得到的原始发送端如果一直保留，result_rx.into_iter() 永远无法通过“所有发送端消失”判断结束。

所以在线程创建完后立刻 drop(result_tx)。此后，只有工作线程持有结果发送端。工作线程退出时，各自发送端被丢弃；最后一个工作线程退出且已有结果被取完后，主线程的结果迭代自然结束。

任务侧采用同样的思路，不过关闭动作是隐式的。job_txs 被整体移进分发线程。分发循环正常结束、提前返回或 panic 时，这个发送端集合都会被丢弃。每个工作线程的 for job in job_rx 随后在队列排空后结束。

这就是所有权参与关闭协议的地方：谁持有发送端，谁就持有“数据流未来还可能继续”的权利。

发送失败时把任务拿回来

分发线程没有对 send(job) 直接 unwrap()。若某个工作线程提前退出，它的接收端会被丢弃，send 返回的错误中包含原来的 Job。

示例把未送达任务作为 Err(Job) 返回。这样协调者至少知道哪条任务没有交付。更完整的系统可以把它重新分配给仍存活的工作线程，但要先回答一个业务问题：失败线程是否可能已经处理任务，只是在发送结果之前退出？

如果答案是“可能”，简单重试可能造成重复执行。任务若会扣款、发邮件或写入外部系统，就需要幂等键、提交记录或其他去重机制。通道只能告诉你消息有没有成功交给接收端，不能证明业务副作用执行了几次。

结果通道断开不等于任务全部成功

result_rx 结束只说明所有结果发送端都消失了，而且已发送结果都取完了。某个工作线程可能正常完成，也可能 panic 后丢掉发送端。

因此示例接收完后仍然逐个 join 工作线程。只有所有句柄都返回 Ok，我们才知道线程入口没有 panic。即便如此，若任务函数返回业务错误，还应让 Completed 携带 Result，由汇总者逐项处理。

这三层状态最好分开：

• 通道关闭描述数据流不会再产生新消息；
• join 描述线程是否正常返回；
• 任务结果描述这项工作在业务上是否成功。

把它们全部压成一次 unwrap()，代码确实短，但你会在最需要诊断时丢掉上下文。

轮转分发也有权衡

示例按 index % worker_count 分发，容易理解，却不考虑任务耗时。如果第一个工作线程连续收到几个特别慢的任务，它的有界队列会先满，分发线程就会在那里等待，即使另外两个工作线程已经空闲。

你可以为任务估算权重、根据工作线程反馈选择空闲者，或者引入真正的多消费者工作队列。每种改进都增加新的状态和失败路径。入门阶段先把简单策略的语义做对，再用数据判断是否值得增加调度复杂度。

关闭与取消不是同一件事

丢掉任务发送端表示“不再有新任务”，工作线程仍会处理缓冲区里已有任务。这属于平滑关闭。

紧急取消的语义不同。你可能希望：

• 立即停止，放弃尚未开始的任务；
• 完成当前任务，但不再领取下一条；
• 继续处理所有已接收任务，只拒绝新任务；
• 等待一个安全提交点再停止。

可以用原子标志或单独的控制通道表达取消请求，但工作线程必须在合适的位置主动检查。Rust 线程没有安全的“从外面强杀”操作，因为线程可能正在持锁、更新文件或维护复杂不变量。强行中断会把资源留在不可预测的状态。

如果任务会长时间运行，就把它拆成可检查取消的阶段。检查太频繁会增加开销，检查太少又会让取消反应迟钝。这个间隔也是业务设计，不是某个 API 的默认答案。

练习：把规则变成判断

把编译器当成严格但负责的搭档

回头看最开始的两个错误，你会发现 Rust 没有禁止并发，它只是不接受含糊的并发：

• 普通线程借用局部数据，生命周期说不清，就移动所有权或改用作用域线程。
• 多个线程同时修改一个值，访问边界说不清，就拆分数据、传消息、加锁或使用经过论证的原子操作。
• 子线程可能失败，就保留句柄并处理两层错误。
• 生产和消费速度不匹配，就选择带容量的通道建立背压。
• 多把锁可能互相等待，就固定顺序并缩小未知等待范围。

真正写多线程程序时，你还是会遇到难受的时刻。一个 Arc<Mutex<...>> 套进泛型后，错误信息可能铺满半个终端；一次偶发死锁也可能让你怀疑整套设计。但别急着用 unsafe、到处 clone，或者加几段 sleep 把错误压下去。

先把问题翻译成三句话：

1. 数据现在归谁？
2. 谁可能同时访问它？
3. 一个参与者停下或失败时，其他参与者靠什么结束？

只要这三句能得到具体答案，类型和同步原语通常会自然收敛。Rust 的编译器有点像那个会在合并请求里追着你问边界条件的同事，确实严格，但它盯住的恰好都是并发程序最容易欠下的债。