跳转到主内容并发:Send、Sync、Arc 和 Mutex | 物联网民工

并发:Send、Sync、Arc 和 Mutex

C 里写线程并发,基本工具很直接:

pthread_create(&tid, NULL, worker, ctx);
pthread_mutex_lock(&lock);
/* modify shared state */
pthread_mutex_unlock(&lock);

这些工具能写出可靠系统,但规则主要靠人维护:

  • ctx 指向的数据能不能跨线程使用
  • 谁拥有这块数据
  • 多个线程是否同时读写
  • 有没有忘记加锁
  • 锁保护的到底是哪几个字段
  • 回调和线程退出时资源是否还活着

Rust 没有让并发自动变简单。它做的是把一部分线程安全要求放进类型系统:

能不能把值移动到另一个线程 -> Send
能不能多个线程共享引用     -> Sync
多线程共享所有权           -> Arc<T>
多线程修改共享状态         -> Mutex<T>

这一篇只讲写命令行工具和库时最常遇到的线程模型,不展开底层内存序和无锁编程。

Send:值能不能跨线程移动

最简单的线程代码:

let data = vec![1_u8, 2, 3];

let handle = std::thread::spawn(move || {
    println!("{}", data.len());
});

handle.join().unwrap();

move 把 data 的所有权移动进新线程。Vec<u8> 可以这样做,因为它是 Send。

Send 的意思是:

这个类型的值可以安全地移动到另一个线程。

大多数普通拥有类型都是 Send,例如 String、Vec<T>、Box<T>,前提是里面的 T 也满足要求。

但不是所有东西都能跨线程移动。比如 Rc<T> 不是线程安全引用计数,不能发到另一个线程:

use std::rc::Rc;

let data = Rc::new(String::from("shared"));

// std::thread::spawn(move || {
//     println!("{data}");
// });

这类代码会被编译器拒绝。C 里类似问题通常是“这个引用计数对象是不是线程安全”,Rust 把它写进类型:跨线程用 Arc<T>,不是 Rc<T>。

Sync:引用能不能被多个线程共享

Sync 的意思是:

如果 &T 可以安全地被多个线程共享,那么 T 是 Sync。

只读共享数据通常没问题:

use std::sync::Arc;

let config = Arc::new(String::from("config"));

let a = Arc::clone(&config);
let b = Arc::clone(&config);

let t1 = std::thread::spawn(move || {
    println!("{a}");
});

let t2 = std::thread::spawn(move || {
    println!("{b}");
});

t1.join().unwrap();
t2.join().unwrap();

String 的只读访问可以多线程共享,Arc<String> 提供线程安全引用计数。

Send 和 Sync 可以先这样分:

Send -> 值能不能搬到另一个线程
Sync -> 引用能不能被多个线程同时共享

大多数时候你不需要手写实现它们。标准库类型和你自己的普通结构体会根据字段自动获得这些能力。真正要手写 unsafe impl Send 或 unsafe impl Sync 的情况很少,通常出现在封装裸指针或 FFI handle 时。

Arc 表达多线程共享所有权

C 里如果多个线程共享同一份配置或状态,常见做法是传同一个指针:

pthread_create(&t1, NULL, worker, shared);
pthread_create(&t2, NULL, worker, shared);

这要求你自己保证 shared 在所有线程结束前都活着。

Rust 里多线程共享所有权用 Arc<T>:

use std::sync::Arc;

let config = Arc::new(Config {
    retry: 3,
});

let worker_config = Arc::clone(&config);

let handle = std::thread::spawn(move || {
    println!("{}", worker_config.retry);
});

handle.join().unwrap();

Arc::clone 不复制底层 Config。它只增加引用计数,让多个 owner 指向同一个值。

不要把 Arc 理解成“自动让内部数据可变”。Arc<T> 只解决多线程共享所有权。它不允许你随便改 T。

Mutex 保护可变共享状态

如果多个线程要修改同一份状态,需要同步。Rust 常见写法是 Arc<Mutex<T>>:

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

let counter = Arc::new(Mutex::new(0_u32));

let mut handles = Vec::new();

for _ in 0..4 {
    let counter = Arc::clone(&counter);

    handles.push(std::thread::spawn(move || {
        let mut value = counter.lock().unwrap();
        *value += 1;
    }));
}

for handle in handles {
    handle.join().unwrap();
}

println!("{}", *counter.lock().unwrap());

这里两层含义要分开:

Arc       -> 多个线程拥有同一份状态
Mutex     -> 同一时间只有一个线程能修改状态

lock() 返回一个 guard。guard 存在期间,锁保持持有;guard 离开作用域时,锁自动释放。

{
    let mut value = counter.lock().unwrap();
    *value += 1;
} // 这里自动 unlock

这和 C 的 pthread_mutex_lock / unlock 不同:Rust 用 guard 和 Drop 把解锁绑定到作用域,减少忘记 unlock 的风险。

锁保护的是数据,不只是代码段

C 里锁经常是一个单独字段:

struct Shared {
    pthread_mutex_t lock;
    int counter;
};

Rust 的 Mutex<T> 把锁和被保护的数据包在一起:

struct Shared {
    counter: Mutex<u32>,
}

你不能在不加锁的情况下直接拿到 counter 的可变引用。要访问里面的数据,必须先 lock():

let mut counter = shared.counter.lock().unwrap();
*counter += 1;

这改变了代码组织方式:锁不只是某段代码前后的操作,而是数据访问权限的一部分。

如果一个结构体里只有少数字段需要并发修改,可以只给这些字段加锁,不必把整个结构体都包进一个大 Mutex。

Mutex poison 不是锁坏了

lock().unwrap() 看起来有点奇怪,因为 lock() 返回 Result。原因是 Rust 的 Mutex 有 poison 机制。

如果某个线程持有锁时 panic,锁会被标记为 poisoned。后续 lock() 会返回错误,提醒你:这份共享状态可能在修改中途停下来了。

简单示例和测试里常写:

let mut value = mutex.lock().unwrap();

生产库代码里可以更明确地处理:

let mut value = mutex.lock().map_err(|_| Error::StatePoisoned)?;

poison 不是说底层锁无法再使用,而是提醒共享数据可能不满足业务不变量。

Arc<Mutex> 不是默认答案

看到并发状态就套 Arc<Mutex<T>>,很容易把问题推迟。

有些场景不需要共享可变状态:

let data = load_input();

let handle = std::thread::spawn(move || {
    process(data)
});

这里直接把所有权移动到线程里就够了。

有些场景只需要共享只读配置:

let config = Arc::new(config);

不需要 Mutex。

有些场景可以通过 channel 传消息,而不是共享状态:

let (tx, rx) = std::sync::mpsc::channel();

线程之间传递任务或结果时,消息队列经常比共享锁更清楚。

先问清楚:

能不能直接移动所有权?
只是只读共享吗?
真的需要多个线程修改同一份状态吗?
消息传递是否更简单?

只有确实需要共享可变状态时,再用 Arc<Mutex<T>>。

原子类型适合简单计数和标志

C 里常用 atomic 做计数、标志位和无锁同步。Rust 也有原子类型:

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

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

但内存序不是入门并发的重点。对多数库和命令行工具来说,先用 Mutex、channel 和线程 join 把模型写清楚。

原子适合很小的共享状态:计数器、停止标志、状态位。不要把复杂结构体的并发一致性寄托在几个 atomic 上,除非你明确知道内存序和不变量。

这一篇先记住什么

C 和 Rust 的并发对应关系可以先这样看:

pthread_create + ctx        -> thread::spawn(move || ...)
共享只读指针                -> Arc<T>
共享可变指针 + mutex        -> Arc<Mutex<T>>
pthread_mutex_lock/unlock   -> Mutex guard + Drop 自动解锁
线程安全引用计数            -> Arc<T>
单线程引用计数              -> Rc<T>,不能跨线程
能移动到线程                -> Send
能被多线程共享引用          -> Sync
简单原子标志/计数           -> Atomic*

Rust 并发的关键不是“没有锁”,而是锁、共享所有权和跨线程移动都进入类型系统。

从 C 迁移时,先少共享:能移动所有权就移动,只读共享用 Arc,确实需要共享可变状态再用 Arc<Mutex<T>>。这样并发边界会比“传一个 void *ctx,大家约定别乱改”清楚得多。

下一步进入 unsafe 和 FFI 边界。前面这些安全抽象解决不了所有底层问题,但它们能让真正需要 unsafe 的代码范围变小。