ShowDocs
搜索文章Ctrl K

外观

← 返回Rust列表
开发实践

Rust 有界并发项目：Send、Sync、Arc 与退出协议 ​

前置：trait 与借用、Result、Cargo 测试。完整项目读取 UTF-8 文本，对每行统计按 Unicode 空白分隔的词数，输出顺序与输入一致。只有标准库；不是 HTTP 服务，不需要外部账号或网络。

Send、Sync 与 Arc ​

Send 允许把值的所有权转移到另一线程，Sync 允许在多线程共享引用，精确定义是 T 为 Sync 当且仅当 &T 为 Send。大多数普通类型由编译器按成员自动推导。Rc 的引用计数不是线程安全的；Arc 的引用计数用原子操作维护，但 Arc<RefCell<T>> 并不因此变成可共享修改。内部修改需要 Mutex/RwLock 或恰当原子变量。

Counter: Send + Sync 让同一个计数策略可以由 Arc 分享给所有线程。输入迭代器用 Mutex 串行领取任务，锁 guard 在执行耗时 count 前释放。锁只保护领取过程，计算可以重叠；若把 count 放在锁内，就会把工作串行化。原子 cancel 只表示“可停止领取”，不发布其他数据，因此 Relaxed 足够；结果通过 join 返回，不用该标志建立内存可见性协议。

以下完整负例执行 rustc --edition=2024 negative/send.rs 应以 E0277 拒绝；不要用 unsafe impl Send 绕过。用 Arc 替换 Rc 后还要确认内部状态满足 Sync。

rs
use std::rc::Rc;
fn main() {
    let value = Rc::new(1);
    std::thread::spawn(move || println!("{value}"));
}

项目契约与文件 ​

这是Rust有界线程保序实验：limit限制线程数量，结果按输入顺序返回；正常创建工作线程后，join等待所有已创建线程结束才交付结果或计数错误。本轮示例未执行，以下测试通过、输出和诊断均为未来自跑预期。

项目目录 examples/rust/concurrency-lab。输入一次读入内存，因此不是恒定内存的流式文件处理器。limit 必须为正；实际线程数为 min(limit, 行数)。输出 Vec 使用 O(n) 内存，空输入成功返回空列表。任一计数错误或 panic 使整体返回 Err，丢弃部分成功结果；返回前 join 所有已创建线程。

取消是合作式的：已进入的同步 count 不会被强制终止，必须自行终止；本例 Words 是有限计算。它不支持超时抢占、阻塞 I/O 中断或业务回滚。若需要这些保证，应改变回调协议、给 I/O 配置超时或隔离进程。panic=abort 会终止进程，不会由 join 转成错误；此处用 Cargo 默认 unwind 配置。线程创建耗尽仍可导致 spawn panic，此时创建循环会展开，已保存的 JoinHandle 被丢弃而分离，因而不属于等待全部线程退出的保证。生产库可用 Builder::spawn 处理 I/O 错误，并在返回前回收此前成功创建的线程。

Cargo.toml：

toml
[package]
name = "concurrency-lab"
version = "0.1.0"
edition = "2024"
rust-version = "1.85"

[dependencies]

src/lib.rs 同时提供接口、借用示例与受约束 unsafe 实验；实际任务算法全部在这里，不用测试替身冒充实现：

rs
use std::sync::{Arc, Mutex, atomic::{AtomicBool, Ordering}};
use std::thread;

pub trait Counter: Send + Sync {
    fn count(&self, text: &str) -> Result<usize, String>;
}

pub struct Words;
impl Counter for Words {
    fn count(&self, text: &str) -> Result<usize, String> {
        Ok(text.split_whitespace().count())
    }
}

pub fn first(text: &str) -> Option<&str> {
    text.lines().next()
}

/// Split a borrowed slice; safe callers cannot create overlapping mutable views.
pub fn split_checked<T>(slice: &mut [T], mid: usize) -> Option<(&mut [T], &mut [T])> {
    if mid > slice.len() { return None; }
    let len = slice.len();
    let ptr = slice.as_mut_ptr();
    // SAFETY: ptr is aligned/non-null even for empty/ZST slices; both ranges lie
    // in the same live allocation and are disjoint. The returned lifetimes are
    // tied to the exclusive input borrow. add(mid) is in-bounds or one-past-end.
    Some(unsafe {
        (std::slice::from_raw_parts_mut(ptr, mid),
         std::slice::from_raw_parts_mut(ptr.add(mid), len - mid))
    })
}

pub fn analyze<C: Counter + 'static>(items: Vec<String>, counter: Arc<C>, limit: usize) -> Result<Vec<usize>, String> {
    if limit == 0 { return Err("limit must be positive".into()); }
    let total = items.len();
    let input = Arc::new(Mutex::new(items.into_iter().enumerate()));
    let cancel = Arc::new(AtomicBool::new(false));
    let mut handles = Vec::new();
    for _ in 0..limit.min(total) {
        let (input, cancel, counter) = (input.clone(), cancel.clone(), counter.clone());
        handles.push(thread::spawn(move || {
            let mut output = Vec::new();
            loop {
                if cancel.load(Ordering::Relaxed) { break; }
                let next = input.lock().map_err(|_| "input lock poisoned".to_string())?.next();
                let Some((index, text)) = next else { break; };
                match counter.count(&text) {
                    Ok(value) => output.push((index, value)),
                    Err(err) => { cancel.store(true, Ordering::Relaxed); return Err(err); }
                }
            }
            Ok(output)
        }));
    }
    let mut output = vec![0; total];
    let mut error = None;
    // Always join all threads, including after error or panic.
    for handle in handles {
        match handle.join() {
            Ok(Ok(values)) => for (index, value) in values { output[index] = value; },
            Ok(Err(err)) => { cancel.store(true, Ordering::Relaxed); error.get_or_insert(err); },
            Err(_) => { cancel.store(true, Ordering::Relaxed); error.get_or_insert("worker panicked".into()); },
        }
    }
    match error { Some(err) => Err(err), None => Ok(output) }
}

src/main.rs：

rs
use concurrency_lab::{Words, analyze};
use std::{env, fs, sync::Arc};

fn run() -> Result<(), String> {
    let mut args = env::args().skip(1);
    let file = args.next().ok_or("usage: concurrency-lab FILE [WORKERS]")?;
    let workers: usize = args.next().unwrap_or("2".into()).parse().map_err(|_| "invalid workers")?;
    if args.next().is_some() { return Err("too many arguments".into()); }
    let text = fs::read_to_string(file).map_err(|err| err.to_string())?;
    let items = text.lines().map(str::to_owned).collect();
    for count in analyze(items, Arc::new(Words), workers)? { println!("{count}"); }
    Ok(())
}
fn main() {
    if let Err(err) = run() { eprintln!("concurrency-lab: {err}"); std::process::exit(1); }
}

运行、错误与调试 ​

在项目目录创建 UTF-8 的 input.txt，内容两行 hello rust 和 one。PowerShell 可用 Set-Content -Encoding utf8 input.txt "hello rustnone"；Bash 用 printf 'hello rust\none\n' > input.txt`。然后两种 Shell 均运行：

sh
cargo test --offline
cargo run --offline -- input.txt 2
cargo run --offline -- input.txt 0
cargo run --offline -- missing.txt 2

第一条测试全部通过；正常进程 stdout 两行 2、1。workers=0 和缺文件都 stderr 报错并退出1，无部分 stdout；非法 UTF-8 同样返回读取错误。CRLF 由 lines 处理，空行计0，中文没有空白时视为一个词；它不是自然语言分词器。RUST_BACKTRACE=1 可辅助分析 panic，PowerShell 用 $env:RUST_BACKTRACE='1'，Bash 用同名环境前缀。不要以启用 backtrace 代替返回 Result。

Rust 的 Future 是可轮询状态机，异步运行时通过 poll 推进；async 不等于创建线程。本项目选择有限 CPU/字符串处理的 std 线程，避免引入与任务无关的运行时。异步锁 guard、跨 await 的借用和阻塞调用有不同约束，不能把这里的线程取消标志当成异步取消协议。

行为测试与练习 ​

Rust错误或panic发生时也必须等待线程退出，不能提前把仍在执行的回调遗留到调用方返回之后；Barrier用于确定性证明这一退出协议。

测试用 Barrier 让两个回调同时在途，核对并发峰值与退出后活跃数；输入长度不同核对保序。错误与 panic 两条路径都实际触发。测试不推断调度顺序，也不依赖短暂 sleep 碰运气。

rs
use concurrency_lab::{analyze, split_checked, Counter, Words, first};
use std::sync::{Arc, Barrier, atomic::{AtomicUsize, Ordering}};

#[test]
fn trait_borrow_and_unsafe_boundary() {
    let input = String::from(" a b\nc ");
    assert_eq!(first(&input), Some(" a b"));
    assert_eq!(Words.count(&input).unwrap(), 3);
    let mut nums = [1, 2, 3];
    let (a,b) = split_checked(&mut nums, 1).unwrap();
    a[0] += 1; b[0] += 2;
    assert_eq!(nums, [2,4,3]);
    assert!(split_checked(&mut nums, 4).is_none());
    assert_eq!(split_checked(&mut nums, 0).unwrap().0.len(), 0);
    assert_eq!(split_checked(&mut nums, 3).unwrap().1.len(), 0);
    assert_eq!(split_checked::<u8>(&mut [], 0).unwrap(), (&mut [][..], &mut [][..]));
    let mut zst = [(); 3];
    let (left, right) = split_checked(&mut zst, 1).unwrap();
    assert_eq!((left.len(), right.len()), (1, 2));
}

struct Measured { active: Arc<AtomicUsize>, peak: Arc<AtomicUsize>, gate: Arc<Barrier> }
impl Counter for Measured {
    fn count(&self, text: &str) -> Result<usize, String> {
        let n = self.active.fetch_add(1, Ordering::SeqCst) + 1;
        self.peak.fetch_max(n, Ordering::SeqCst);
        self.gate.wait();
        self.active.fetch_sub(1, Ordering::SeqCst);
        Ok(text.len())
    }
}
#[test]
fn bound_order_empty_and_invalid() {
    let active = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let counter = Arc::new(Measured { active: active.clone(), peak: peak.clone(), gate: Arc::new(Barrier::new(2)) });
    assert_eq!(analyze(vec!["aaa".into(), "x".into(), "abcd".into(), "xx".into()], counter, 2).unwrap(), vec![3,1,4,2]);
    assert_eq!(peak.load(Ordering::SeqCst), 2);
    assert_eq!(active.load(Ordering::SeqCst), 0);
    assert_eq!(analyze(vec![], Arc::new(Words), 2).unwrap(), vec![]);
    assert!(analyze(vec![], Arc::new(Words), 0).is_err());
}

struct Fails;
impl Counter for Fails {
    fn count(&self, text: &str) -> Result<usize, String> {
        if text == "bad" { Err("bad input".into()) } else { Ok(1) }
    }
}
struct Panics;
impl Counter for Panics {
    fn count(&self, _: &str) -> Result<usize, String> { panic!("worker panicked") }
}
#[test]
fn error_and_panic_are_joined() {
    assert!(analyze(vec!["bad".into(), "ok".into()], Arc::new(Fails), 2).unwrap_err().contains("bad input"));
    assert!(analyze(vec!["ok".into()], Arc::new(Panics), 2).unwrap_err().contains("panicked"));
}

struct UnwindMeasured { active: Arc<AtomicUsize>, gate: Arc<Barrier> }
struct ActiveGuard(Arc<AtomicUsize>);
impl Drop for ActiveGuard {
    fn drop(&mut self) { self.0.fetch_sub(1, Ordering::SeqCst); }
}
impl Counter for UnwindMeasured {
    fn count(&self, text: &str) -> Result<usize, String> {
        self.active.fetch_add(1, Ordering::SeqCst);
        let _guard = ActiveGuard(self.active.clone());
        self.gate.wait();
        if text == "panic" { panic!("failure cleanup"); }
        if text == "bad" { return Err("failure cleanup".into()); }
        Ok(text.len())
    }
}
#[test]
fn failed_work_leaves_no_live_callback() {
    for failure in ["bad", "panic"] {
        let active = Arc::new(AtomicUsize::new(0));
        let counter = Arc::new(UnwindMeasured { active: active.clone(), gate: Arc::new(Barrier::new(2)) });
        assert!(analyze(vec![failure.into(), "ok".into()], counter, 2).is_err());
        assert_eq!(active.load(Ordering::SeqCst), 0);
    }
}

练习 ​

① 将计数策略换成 Unicode 标量数，应改调度器吗？② 如果所有线程在 count 内永久阻塞，join 能否保证超时退出？③ 为什么不能一遇 Err 就返回而丢弃其他 handles？④ 输出长度与完成顺序不同该怎样维护？

参考解法 ​

① 新建结构体实现 Counter，返回 text.chars().count()，把 Arc 交给 analyze，调度器不变。② 不能；join 等待线程结束，取消标志只在工作边界检查。③ Drop JoinHandle 会分离线程，回调可能在函数返回后继续访问共享资源；应先完成退出协议。④ 领取时保存序号，回收时写回序号位置；不要将到达顺序当作输入顺序。

来源 ​

核对日期2026-10-17：Send/Sync Book、Arc、JoinHandle、官方中文学习入口。这些接口定义了能力边界；本站执行器及测试独立实现。

继续阅读