mcpField
R
rust-concurrency

rust-concurrency

github.com/rewrite-rs其他1039 次安装10 周前更新
代码仓库

关于 rust-concurrency

面向 Rust 开发者的并发技能，介绍如何根据负载选择数据并行、作用域线程、通道或共享状态，并说明同步与安全方面的注意事项。

主要功能

  1. 01按负载选择并发模型
  2. 02使用 Rayon 实现数据并行
  3. 03通过作用域线程借用栈数据
  4. 04用通道转移所有权
  5. 05锁与原子排序指导

适用场景

  1. 01并行处理计算量较大的独立任务
  2. 02在线程间共享栈数据
  3. 03无需共享锁地传递任务

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/rewrite-rs/skills --skill rust-concurrency

安装后在 Claude Code 中输入/rust-concurrency确认已加载。

附带文件agents/openai.yaml

详细介绍

Rust 并发

工作负载的形态决定采用哪种模型。这项技能负责线程相关的选择，并在任务并发开始处止步——运行时、执行器、spawn_blocking 和取消属于 /async-rust。

工作负载的形态决定模型

  • 对许多独立项执行相同操作：数据并行，使用 rayon。
  • 大多处于等待状态的独立工作单元：任务并发，参见 /async-rust，不在此处讨论。
  • 多个线程读写的状态：共享状态——最后才考虑这种方式，因为三种方式中只有它可能导致死锁。

数据并行

par_iter() 是对现有迭代器链的一处改动——这正是编写该迭代器链的价值所在。前提是：每项工作的计算量必须足够大，值得承担调度开销。对一百万个开销很小的闭包使用 par_iter，通常会更慢，也是最常见的失望之处。

作用域线程

std::thread::scope 允许线程借用栈数据，因为作用域会保证在线程栈展开前完成连接——这就去掉了原本会迫使你使用 Arc 的 'static 限制：

fn sum_halves(data: &[u64]) -> u64 {
    let (left, right) = data.split_at(data.len() / 2);
    std::thread::scope(|s| {
        let a = s.spawn(|| left.iter().sum::<u64>());
        let b = s.spawn(|| right.iter().sum::<u64>());
        a.join().unwrap() + b.join().unwrap()
    })
}

在使用 Arc 或克隆来创建第二个线程之前，先考虑作用域线程。

用通道移交数据

通道会随消息转移所有权——无需持有锁，也没有中毒问题。发送端丢弃就是关闭信号：recv 会报告连接断开。有界与无界的区别在于背压：无界通道会悄无声息地把吞吐量问题变成内存问题。

共享状态

默认使用 Mutex；只有在确实以读取为主时，RwLock 才有优势，而且它可能让写入者饥饿。持有锁的线程发生 panic 后，Mutex 会返回 Err——传播这个错误，或说明理由后使用 into_inner()，绝不要习惯性地调用 unwrap。尽可能缩短持有 guard 的作用域，绝不要在可能再次加锁的代码调用期间持有它。

原子操作与内存顺序

规则如下：对于无人据此进行同步的计数器，使用 Relaxed；对于由一个线程发布、另一个线程观察的成对移交，使用 Acquire/Release；不确定时使用 SeqCst，前提是开销可以接受。最常出错的是这一对：

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

fn main() {
    let value = Arc::new(AtomicU64::new(0));
    let ready = Arc::new(AtomicBool::new(false));
    let (v, r) = (Arc::clone(&value), Arc::clone(&ready));
    let publisher = std::thread::spawn(move || {
        v.store(42, Ordering::Relaxed);
        r.store(true, Ordering::Release);
    });
    while !ready.load(Ordering::Acquire) {
        std::thread::yield_now();
    }
    assert_eq!(value.load(Ordering::Relaxed), 42);
    publisher.join().unwrap();
}

Release 存储会确保对 value 的写入先于 ready 翻转，而 Acquire 加载会与之配对。如果无法用两句话写清内存顺序的理由，那么这段代码应该使用 Mutex。

手动实现 Send/Sync 是一项不安全的声明

unsafe impl Send for T 声明了编译器无法证明的事实；// SAFETY: 注释必须给出实际论证。健全性审查属于 /unsafe-rust——这项技能只要求写出这项声明。

使用 thread_local!，而非 static mut

在当前版本中，static mut 实际上无法使用，而且共享它从来都不健全；每个线程各自持有状态时，应使用 thread_local!。显式传递状态胜过把它藏在全局变量中——全局变量的生命周期没有人明确写出。

延后讨论

任务并发、执行器、spawn_blocking 和取消属于 /async-rust。是否应该共享状态属于 /ownership-not-clone。unsafe impl 的健全性属于 /unsafe-rust。并行版本的吞吐量属于 /rust-performance。

验证

cargo test                    # including the ignored-by-default stress tests, if the repo has any
cargo clippy --all-targets    # at the level the repo configures
cargo miri test               # where atomics or unsafe Send/Sync are involved

Clippy 会采用目标仓库配置的 lint 级别；仅当仓库完全没有配置 lint 时，才使用 -D warnings 作为后备方案。无锁结构和不简单的内存顺序论证应使用 loom——它会穷举交错执行，而不是寄希望于调度器恰好选中出错的情况；它在自己的 cfg 下运行，且模型必须足够小。

评论

支持 Markdown