吃透 Rust Send 和 Sync：并发编程的基石，案例说话

并发编程是现代软件开发中的一个重要组成部分。Rust 语言以其安全性、高效性和零成本抽象而闻名，特别是在并发编程方面，它通过 Send 和 Sync 这两个 trait 提供了一套强大的安全机制。但这两个 trait 经常让 Rust 新手感到困惑。本文将深入剖析 Send 和 Sync 的作用和区别，并结合实际例子，让你彻底理解如何在 Rust 并发编程中正确使用它们。

1. 什么是 Send 和 Sync？

在 Rust 中，Send 和 Sync 是两个标记 trait（marker traits），它们没有任何方法。它们的作用是指示类型在并发上下文中的安全属性。

• Send: 如果一个类型 T 实现了 Send trait，则意味着 T 的值可以在线程之间安全地转移所有权。换句话说，你可以将 T 的值从一个线程移动到另一个线程。
• Sync: 如果一个类型 T 实现了 Sync trait，则意味着 T 可以安全地在多个线程之间共享（通过引用）。这意味着多个线程可以同时访问 &T 而不会导致数据竞争。

简单来说：

• Send 关系到所有权的转移。
• Sync 关系到共享访问。

2. 默认实现

Rust 编译器会自动为满足特定条件的类型实现 Send 和 Sync。一般来说：

• 所有基本类型（如 i32、f64、bool、char 等）都实现了 Send 和 Sync。 这是因为它们在内存中是简单、可复制的，不存在数据竞争的风险。
• 如果一个类型的所有字段都实现了 Send，那么该类型自动实现 Send。
• 如果一个类型的所有字段都实现了 Sync，那么该类型自动实现 Sync。

需要注意的是，如果类型包含原始指针（raw pointer，如 *mut T 和 *const T），则该类型不会自动实现 Send 和 Sync。这是因为原始指针是不安全的，Rust 编译器无法保证它们在并发上下文中的正确使用。

3. 何时需要手动实现 Send 和 Sync？

大多数情况下，你不需要手动实现 Send 和 Sync，因为 Rust 编译器会自动推导。但是，在以下情况下，你可能需要考虑手动实现或使用 unsafe 代码来处理：

• 类型包含原始指针，并且你确定你的使用方式是线程安全的。 例如，你可能使用原子操作来保护原始指针指向的数据。
• 你正在编写底层代码，需要与外部 C 代码交互。
• 你正在使用一些不安全的 Rust 特性，并且需要手动保证线程安全。

在这种情况下，你需要非常小心，确保你的代码确实是线程安全的，否则可能会导致数据竞争和未定义行为。

4. 常见类型与 Send/Sync

理解常见类型如何与 Send 和 Sync 交互非常重要：

• Arc<T>: Arc<T>（原子引用计数）是线程安全的共享所有权智能指针。如果 T 实现了 Send 和 Sync，那么 Arc<T> 也实现了 Send 和 Sync。这使得在多个线程之间共享数据变得容易。
• Mutex<T>: Mutex<T> 提供互斥锁，用于保护共享数据。即使 T 没有实现 Sync，Mutex<T> 也可以安全地在线程之间共享。这是因为 Mutex<T> 提供了内部可变性，并通过锁机制来保证数据访问的互斥性。只有当 T 实现了 Send，Mutex<T> 才能实现 Send。
• RwLock<T>: RwLock<T> 提供读写锁，允许多个读者同时访问数据，但只允许一个写者独占访问。类似于 Mutex<T>，RwLock<T> 也提供了内部可变性，并且需要 T 实现 Send 才能实现 Send。
• Rc<T>: Rc<T>（引用计数）是单线程的引用计数智能指针。它没有实现 Send 和 Sync，因此不能在线程之间安全地共享。如果你需要在多线程环境中使用引用计数，应该使用 Arc<T>。
• Cell<T> 和 RefCell<T>: Cell<T> 和 RefCell<T> 提供内部可变性，但它们没有实现 Sync，因此不能在线程之间安全地共享。这是因为它们没有提供任何同步机制来防止数据竞争。

5. 实际案例分析

5.1 多线程数据共享

假设我们需要创建一个多线程程序，其中多个线程需要访问和修改同一个计数器。我们可以使用 Arc<Mutex<T>> 来实现：

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

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

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

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

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

在这个例子中：

• counter 是一个 Arc<Mutex<i32>>，它允许多个线程共享对计数器的所有权。
• Mutex<i32> 提供了互斥锁，确保只有一个线程可以同时访问和修改计数器。
• Arc::clone(&counter) 创建一个新的 Arc 实例，增加引用计数。每个线程都拥有 counter 的一个副本。
• counter.lock().unwrap() 获取互斥锁，阻塞直到锁可用。unwrap() 用于处理锁获取失败的情况（例如，当持有锁的线程 panic 时）。
• *num += 1 增加计数器的值。
• 当线程完成时，锁会自动释放。

5.2 线程池

线程池是一种常见的并发模式，它可以重用线程来执行多个任务。我们可以使用 Send 和 Sync 来创建一个简单的线程池：

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

trait FnBox: Send + 'static {
    fn call_box(self: Box<Self>);
}

impl<F: FnOnce() + Send + 'static> FnBox for F {
    fn call_box(self: Box<Self>) {
        (*self)()
    }
}

type Job = Box<dyn FnBox + Send>;

struct Worker {
    id: usize,
    thread: Option<thread::JoinHandle<()>>,
}

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        let thread = thread::spawn(move || loop {
            let message = receiver.lock().unwrap().recv();

            match message {
                Ok(job) => {
                    println!("Worker {} got a job; executing.", id);

                    job.call_box();
                }
                Err(_) => {
                    println!("Worker {} disconnected; terminating.", id);
                    break;
                }
            }
        });

        Worker {
            id,
            thread: Some(thread),
        }
    }
}

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

impl ThreadPool {
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        let (sender, receiver) = mpsc::channel();

        let receiver = Arc::new(Mutex::new(receiver));

        let mut workers = Vec::with_capacity(size);

        for id in 0..size {
            workers.push(Worker::new(id, Arc::clone(&receiver)));
        }

        ThreadPool {
            workers,
            sender,
        }
    }

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);

        self.sender.send(job).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        println!("Sending terminate message to all workers.");

        for worker in &mut self.workers {
            println!("Shutting down worker {}", worker.id);

            if let Some(thread) = worker.thread.take() {
                thread.join().unwrap();
            }
        }
    }
}


fn main() {
    let pool = ThreadPool::new(4);

    for i in 0..10 {
        pool.execute(move || {
            println!("Task {} is running on thread {:?}", i, thread::current().id());
            thread::sleep(std::time::Duration::from_secs(1));
        });
    }

    println!("All tasks submitted.");
    //drop(pool);
}

在这个例子中：

• ThreadPool 包含一个 Worker 向量和一个 mpsc::Sender<Job>，用于将任务发送给 worker 线程。
• Worker 结构体包含一个线程和一个 ID。
• mpsc::channel() 创建一个多生产者、单消费者的通道。sender 用于发送任务，receiver 用于接收任务。
• Arc<Mutex<mpsc::Receiver<Job>>> 允许多个 Worker 线程共享对 receiver 的访问。
• execute 方法将一个闭包（实现了 FnOnce() + Send + 'static）包装成一个 Job，并通过 sender 发送给 worker 线程。
• FnBox trait 是一个 trait object，用于在线程之间传递闭包。
• Drop trait 用于在 ThreadPool 销毁时，安全地关闭所有 worker 线程。

关键点：

• 闭包必须实现 Send trait，才能在线程之间传递。
• Arc<Mutex<T>> 用于在多个线程之间共享数据。
• mpsc::channel() 用于在线程之间传递消息。

5.3 何时可能出现问题？

考虑以下代码：

use std::thread;

#[derive(Debug)]
struct NotThreadSafe {
    value: i32,
}

fn main() {
    let data = NotThreadSafe { value: 5 };

    thread::spawn(move || {
        println!("{:?}", data);
    }).join().unwrap();
}

这段代码会编译失败，因为 NotThreadSafe 没有实现 Send trait。编译器会提示错误信息，告诉你 NotThreadSafe 不能在线程之间传递。

要解决这个问题，你需要确保 NotThreadSafe 中的所有字段都实现了 Send 和 Sync。如果 NotThreadSafe 包含原始指针或其他不安全的类型，你可能需要使用 unsafe 代码来手动实现 Send 和 Sync，并确保你的实现是线程安全的。但是，除非你非常清楚自己在做什么，否则应该避免使用 unsafe 代码。

6. 总结

Send 和 Sync 是 Rust 并发编程中非常重要的概念。理解它们的作用和区别，可以帮助你编写安全、高效的并发代码。记住以下几点：

• Send 关系到所有权的转移。
• Sync 关系到共享访问。
• Rust 编译器会自动为满足特定条件的类型实现 Send 和 Sync。
• 在大多数情况下，你不需要手动实现 Send 和 Sync。
• 使用 Arc<Mutex<T>> 和 RwLock<T> 来安全地共享数据。
• 避免在多线程环境中使用 Rc<T>、Cell<T> 和 RefCell<T>。
• 除非你非常清楚自己在做什么，否则应该避免使用 unsafe 代码。

通过掌握 Send 和 Sync，你可以更好地利用 Rust 的并发特性，构建高性能、高可靠性的应用程序。并发编程之路漫漫，希望本文能为你提供一个坚实的起点。