Rust中无需async的多线程计算结果同步：替代RwLock的方案

我希望将CPU密集型计算卸载到线程中，同时获取该计算的引用，以便后续代码中获取结果——类似Future但无需async。期望实现如下逻辑：

let future_result: MyComputationFuture<T> = start_crunching_numbers(params);

// ...执行其他操作，可克隆future_result

let result: T = future_result.await_computation(); // 阻塞调用，可多次执行！

补充需求：future_result需支持多线程克隆，即实现Sync；.await_computation()会在多线程中被多次调用，无法预知首次调用时机，所有获取结果的尝试需阻塞直到计算完成。

我的首次尝试是使用RwLock，简化代码如下：

let resource = std::sync::RwLock::new(0);

{
    // 立即加写锁直到计算完成
    let mut write_lock = resource.write().unwrap();
    std::thread::spawn(move || {
        // 模拟长时间计算
        *write_lock = 42;
    })
};

// 获取结果，未就绪则挂起线程
let result = *resource.read().unwrap();

但RwLockWriteGuard不满足Send trait，编译报错：

error[E0277]: `std::sync::RwLockWriteGuard<'_, i32>` cannot be sent between threads safely

若使用带send_guard标志的parking_lot::RwLock，则出现生命周期错误：

error[E0597]: `resource` does not live long enough
let mut write_lock = resource.write();
                     ^^^^^^^^^^^^^^^^ borrowed value does not live long enough

方案1：标准库原生实现——Arc<Mutex<Option<T>>> + Condvar #

无需额外依赖，通过互斥锁保护计算状态，条件变量通知所有等待线程计算完成，完美适配多线程克隆、多次阻塞获取的需求。

use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;

// 自定义Future类型，基于Arc共享线程安全状态
struct MyComputationFuture<T> {
    state: Arc<(Mutex<Option<T>>, Condvar)>,
}

// 实现Clone，依托Arc的克隆能力
impl<T> Clone for MyComputationFuture<T> {
    fn clone(&self) -> Self {
        Self {
            state: self.state.clone(),
        }
    }
}

// 自动推导Sync/Send（Arc、Mutex、Condvar均为线程安全类型）
unsafe impl<T: Send + Sync> Sync for MyComputationFuture<T> {}
unsafe impl<T: Send> Send for MyComputationFuture<T> {}

impl<T> MyComputationFuture<T> {
    // 阻塞等待结果，支持多线程重复调用
    fn await_computation(&self) -> T
    where
        T: Clone,
    {
        let (mutex, condvar) = &*self.state;
        let mut guard = mutex.lock().unwrap();
        
        // 循环等待避免虚假唤醒，直到结果就绪
        while guard.is_none() {
            guard = condvar.wait(guard).unwrap();
        }
        
        guard.as_ref().unwrap().clone()
    }
}

// 启动CPU密集型计算的入口函数
fn start_crunching_numbers(params: u32) -> MyComputationFuture<u32> {
    let state = Arc::new((Mutex::new(None), Condvar::new()));
    let state_clone = state.clone();
    
    thread::spawn(move || {
        // 模拟长时间计算
        thread::sleep(Duration::from_secs(2));
        let result = params * 2;
        
        // 计算完成后更新状态并通知所有等待线程
        let (mutex, condvar) = &*state_clone;
        let mut guard = mutex.lock().unwrap();
        *guard = Some(result);
        condvar.notify_all();
    });
    
    MyComputationFuture { state }
}

// 使用示例
fn main() {
    let future = start_crunching_numbers(21);
    
    // 克隆future到其他线程
    let future_clone = future.clone();
    thread::spawn(move || {
        let result = future_clone.await_computation();
        println!("线程1获取结果: {}", result);
    });
    
    // 主线程执行其他操作
    println!("主线程执行其他任务...");
    
    // 主线程等待结果
    let result = future.await_computation();
    println!("主线程获取结果: {}", result);
}

方案2：高性能实现——parking_lot::RwLock + 状态枚举 #

引入parking_lot依赖后，其RwLock支持多读不互斥，性能优于标准库RwLock，适合高并发场景下多次获取结果的需求。

首先在Cargo.toml添加依赖：

parking_lot = "0.12"

实现代码：

use parking_lot::{RwLock, RwLockReadGuard};
use std::thread;
use std::time::Duration;

// 用枚举标记计算状态
enum ComputationState<T> {
    Pending,
    Ready(T),
}

struct MyComputationFuture<T> {
    state: Arc<RwLock<ComputationState<T>>>,
}

impl<T> Clone for MyComputationFuture<T> {
    fn clone(&self) -> Self {
        Self {
            state: self.state.clone(),
        }
    }
}

unsafe impl<T: Send + Sync> Sync for MyComputationFuture<T> {}
unsafe impl<T: Send> Send for MyComputationFuture<T> {}

impl<T> MyComputationFuture<T> {
    fn await_computation(&self) -> &T {
        loop {
            let guard = self.state.read();
            match &*guard {
                ComputationState::Ready(result) => return result,
                ComputationState::Pending => {
                    // 释放读锁让出CPU，避免空转
                    drop(guard);
                    thread::yield_now();
                }
            }
        }
    }
}

fn start_crunching_numbers(params: u32) -> MyComputationFuture<u32> {
    let state = Arc::new(RwLock::new(ComputationState::Pending));
    let state_clone = state.clone();
    
    thread::spawn(move || {
        thread::sleep(Duration::from_secs(2));
        let result = params * 2;
        
        // 仅一次写锁更新状态
        let mut guard = state_clone.write();
        *guard = ComputationState::Ready(result);
    });
    
    MyComputationFuture { state }
}

// 使用示例
fn main() {
    let future = start_crunching_numbers(21);
    let future_clone = future.clone();
    
    thread::spawn(move || {
        let result = future_clone.await_computation();
        println!("线程1获取结果: {}", result);
    });
    
    println!("主线程执行其他任务...");
    
    let result = future.await_computation();
    println!("主线程获取结果: {}", result);
}

方案3：极简实现——Arc<OnceLock<T>>（Rust 1.70+） #

标准库OnceLock保证值仅初始化一次，结合线程yield实现等待逻辑，代码最简洁。

use std::sync::Arc;
use std::sync::OnceLock;
use std::thread;
use std::time::Duration;

struct MyComputationFuture<T> {
    cell: Arc<OnceLock<T>>,
}

impl<T> Clone for MyComputationFuture<T> {
    fn clone(&self) -> Self {
        Self {
            cell: self.cell.clone(),
        }
    }
}

unsafe impl<T: Send + Sync> Sync for MyComputationFuture<T> {}
unsafe impl<T: Send> Send for MyComputationFuture<T> {}

impl<T> MyComputationFuture<T> {
    fn await_computation(&self) -> &T {
        // 循环等待直到OnceLock被初始化
        while self.cell.get().is_none() {
            thread::yield_now();
        }
        self.cell.get().unwrap()
    }
}

fn start_crunching_numbers(params: u32) -> MyComputationFuture<u32> {
    let cell = Arc::new(OnceLock::new());
    let cell_clone = cell.clone();
    
    thread::spawn(move || {
        thread::sleep(Duration::from_secs(2));
        let result = params * 2;
        // set方法仅成功一次，重复调用无影响
        let _ = cell_clone.set(result);
    });
    
    MyComputationFuture { cell }
}

// 使用示例
fn main() {
    let future = start_crunching_numbers(21);
    let future_clone = future.clone();
    
    thread::spawn(move || {
        let result = future_clone.await_computation();
        println!("线程1获取结果: {}", result);
    });
    
    println!("主线程执行其他任务...");
    
    let result = future.await_computation();
    println!("主线程获取结果: {}", result);
}

初始尝试失败原因 #

• 标准库RwLockWriteGuard未实现Send，它绑定了原RwLock的生命周期，无法跨线程传递；
• 即使使用parking_lot的可发送Guard，栈上的RwLock生命周期短于线程，直接传递Guard会导致悬垂引用。正确做法是将共享状态放入Arc，让线程持有Arc克隆，在线程内部获取锁。

内容的提问来源于stack exchange，提问作者aedm