rust-concurrency

Rust 并发

基于标准库的 std::thread、std::sync 和 std::sync::atomic 模块，以及 Async Book。适用于设计、调试、负载测试或审查 Rust 线程与异步代码；取消、任务所有权、锁的作用域、运行时规模、队列、过载管理、消息传递；简单的所有权问题交给 rust-stable，unsafe 不变式交给 rust-unsafe-ffi。

能力边界

✅ 优势

1. 操作系统线程（thread::spawn、Builder、join、作用域线程、move 闭包）
2. 同步原语（Mutex、RwLock、Barrier、Condvar、OnceLock、LazyLock）
3. 原子类型（AtomicBool/Isize/Usize、load/store/fetch_add/swap/compare_exchange、Ordering）
4. 通道（mpsc：多生产者单消费者、Receiver、Sender）
5. Send / Sync trait 系统（自动派生和手动实现）
6. 使用 Future trait 的 async/await 语法
7. Tokio 运行时（tokio::main、tokio::spawn、select!、JoinSet）
8. 异步 I/O 基础（tokio::fs、tokio::net、tokio::io）
9. 有界队列、背压、慢速消费者、并发限制和过载策略
10. 任务监管、连接生命周期、取消安全性和优雅关闭
11. CPU 密集型数据并行和专用 Rayon 线程池
12. Crossbeam 通道、队列、工作窃取双端队列和作用域线程
13. 读密集型快照、分片映射、缓存，以及经测量证明合理时采用的替代锁
14. Loom 模型检查和 Tokio 运行时诊断

⚠️ 前置条件

1. 理解 Rust 所有权模型（rust-stable）

❌ 不适用的场景

1. 并发执行 unsafe 代码 → 使用 rust-unsafe-ffi 技能
2. 基础所有权/借用 → 使用 rust-stable 技能

适用场景

• “使用多个线程处理数据”
• “如何编写 async/await”
• “Tokio 运行时的用法”
• “线程间共享数据”
• “避免数据竞争”
• “高并发服务中的速率限制和优雅关闭”
• “Tokio 通道积压或慢速消费者”

数据隐私

此技能不会收集、存储或传输任何用户数据。

---

I. 操作系统线程

use std::thread;

let handle = thread::spawn(move || {
    println!("Hello from thread!");
});
handle.join().unwrap();

// Thread with configuration
let builder = thread::Builder::new()
    .name("worker".into())
    .stack_size(1024 * 1024);
let handle = builder.spawn(move || { /* ... */ }).unwrap();

// scoped threads (1.63+)
let mut v = vec![1, 2, 3];
thread::scope(|s| {
    s.spawn(|| {
        v.push(4); // borrow, no move required
    });
});
println!("{v:?}"); // v remains usable

II. 同步原语

use std::sync::{Arc, Mutex, RwLock, Barrier, OnceLock, LazyLock};

// Mutex (mutual exclusion lock)
let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];

for _ in 0..10 {
    let counter = Arc::clone(&counter);
    handles.push(thread::spawn(move || {
        let mut num = counter.lock().unwrap();
        *num += 1;
    }));
}

// RwLock (read-write lock)
let data = Arc::new(RwLock::new(vec![1, 2, 3]));
{
    let read = data.read().unwrap();
    assert_eq!(read.len(), 3);
} // Drop the read guard before taking the write lock.
data.write().unwrap().push(4);

// OnceLock (thread-safe lazy initialization)
static CONFIG: OnceLock<String> = OnceLock::new();
let config = CONFIG.get_or_init(|| load_config());

// LazyLock
static CACHE: LazyLock<HashMap<String, Data>> = LazyLock::new(HashMap::new);

III. 原子操作

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

static COUNTER: AtomicU64 = AtomicU64::new(0);
COUNTER.fetch_add(1, Ordering::SeqCst);

static READY: AtomicBool = AtomicBool::new(false);
READY.store(true, Ordering::Release);
let ready = READY.load(Ordering::Acquire);

// Ordering levels
// Relaxed — no ordering guarantees (only atomicity)
// Release — write visibility
// Acquire — read visibility
// AcqRel — both reads and writes visible
// SeqCst — global sequential order (strongest, but not automatically default; explicit Ordering required for atomic operations)

IV. 通道

use std::sync::mpsc;

let (tx, rx) = mpsc::channel();
thread::spawn(move || {
    tx.send(1).unwrap();
    tx.send(2).unwrap();
});
for received in rx {
    println!("Got: {received}");
}

// Multi-producer scenario
let (tx, rx) = mpsc::channel();
let tx1 = tx.clone();

V. async/await

use tokio::time;

async fn do_work(id: u32) -> &'static str {
    time::sleep(time::Duration::from_secs(1)).await;
    println!("Task {id} done");
    "ok"
}

#[tokio::main]
async fn main() {
    // Concurrent execution
    let (r1, r2) = tokio::join!(do_work(1), do_work(2));

    // select!
    tokio::select! {
        result = do_work(1) => println!("task1: {result}"),
        result = do_work(2) => println!("task2: {result}"),
    }

    // tokio::spawn
    let handle = tokio::spawn(do_work(3));
    handle.await.unwrap();
}

VI. Send / Sync

// T is Send if its ownership can be transferred across threads
// &T is Sync if it can be shared references across threads

// Types that are both Send + Sync: Arc<Mutex<T>>, i32, &'static str
// !Send types: Rc<T>, *const T
// !Sync types: RefCell<T>, Cell<T>

// Manual implementations are unsafe contracts. Do not add them merely to
// satisfy a compiler error; prove aliasing, lifetime, and thread-safety first.

VII. 选择执行模型

工作负载	默认起点	避免
大量由就绪事件驱动的网络操作	使用有界准入的 Tokio 任务	为无界输入各自创建任务或缓冲区
相互独立的 CPU 密集型项目	Rayon 并行迭代器或专用线程池	在 Tokio 工作线程上执行长时间 CPU 密集型工作
阻塞式文件系统、FFI 或旧式 API	有界的 spawn_blocking 提交或专用线程池	将 Tokio 的阻塞队列当作背压机制
同步 MPMC 消息传递或工作窃取	Crossbeam 通道、队列或双端队列	未经测量就选择无锁数据结构
临界区较短的小型共享状态	std::sync 锁	在 .await 或回调期间持有锁守卫
以读取为主的不可变快照	经分析后使用 ArcSwap	为每个读密集型值都使用并发映射
共享的键控可变状态	经竞争测试后采用分片所有权或 DashMap	在缺少原子性设计时执行多键操作
具有过期机制的并发缓存	使用明确容量和淘汰策略的 Moka	将无界映射称为缓存

Tokio 主要用于 I/O 并发；Rayon 用于 CPU 并行。混合使用时，需要明确交接方式、独立的并发限制和关闭时的所有权管理。引入第三方原语前，请阅读 并发工具选择。

工作流

1. 对工作负载分类 —— 在选择运行时或原语之前，区分由就绪状态驱动的 I/O、CPU 并行处理、阻塞调用、同步和持久化消息传递。
2. 制定并发预算 —— 定义最大连接数、执行中的任务数、队列容量、项目大小、超时时间、内存、CPU 池和关闭期限。
3. 确定状态所有权 —— 优先采用分区所有权或单写入者所有权；只有在明确规定原子性和锁顺序的情况下才共享状态。
4. 选择通信语义 —— 有意选择有界点对点、请求/回复、最新值、有损广播或持久化重放；明确规定队列已满和接收方滞后的处理方式。
5. 监督执行 —— 保留任务或线程句柄、传播失败、隔离 panic、防止产生孤儿工作，并定义调用方取消行为。
6. 设计优雅关闭 —— 停止接收新任务、关闭生产者、发布取消信号、在期限内等待任务完成、刷新必需状态，并返回尚未解决的失败。
7. 先测量，再调优 —— 记录吞吐量、p50/p95/p99 延迟、队列深度、饱和度、任务轮询时间、唤醒次数、锁等待时间、CPU、内存分配和 RSS。
8. 验证模型 —— 测试过载和取消，使用 Loom 检验小型同步状态机，并使用 tokio-console 或 tracing 诊断运行时停滞。阅读 并发测试与诊断。

注意事项

1. Mutex::lock() 会返回 MutexGuard；为避免死锁，丢弃它之前不要执行 await
2. tokio::spawn 的 Future 必须同时满足 Send 和 'static；非 Send 引用会导致编译错误
3. 异步闭包捕获所有权的方式与普通闭包不同 —— 显式使用 move 关键字来转移状态
4. select! 分支中被取消的 Future 在丢弃前不会直接执行清理逻辑
5. Atomic Ordering 不是关系语义；误用 Relaxed 可能导致意外的内存排序问题
6. 广播滞后不同于正常成功路径；必须选择丢弃消息、重建快照、断开慢速消费者，或持久化重放事件
7. max_blocking_threads 仅限制阻塞线程的数量，不会对提交队列提供背压；高成本任务需要使用 Semaphore 或有界队列
8. JoinSet 按完成顺序返回结果；如果 API 要求保持输入顺序，则需传递索引，以便在聚合时恢复顺序
9. DashMap、parking_lot、ArcSwap 和无锁队列不仅会改变性能，也会改变语义；基准测试不能替代不变量审查
10. Loom 只能检测通过 Loom 感知类型执行的同步操作，而且可能出现状态空间爆炸；应让模型保持小巧、确定
11. Rayon 工作可能在异步调用方结束后继续运行，除非明确设计取消机制和线程池所有权

按需资源

• 并发示例
• 类型与工具速查
• 并发工具选择：在选择 Tokio、Rayon、Crossbeam、锁、分片映射、快照或缓存时阅读。
• 并发测试与诊断：在验证同步正确性、诊断运行时停滞，或对过载和关闭进行负载测试时阅读。
• 生产级异步服务模式：在设计 actor、背压、慢速消费者、任务监督、运行时配置和关闭协议时阅读。
• examples/golden-threads/：由 CI 构建的作用域线程示例

Official References

• std::thread Documentation
• std::sync Documentation
• std::sync::atomic Documentation
• Async Book
• Tokio Guide
• Rayon
• Crossbeam
• Loom