如何实现带超时的Mutex锁？适配Modem AT命令线程同步需求

我希望调制解调器AT命令写入线程仅在读取线程检测到“OK”或“ERROR”时，才向调制解调器的/dev/ttyUSB3写入命令。这乍一看适合使用Mutex<()>，但我有额外需求：若读取线程3秒内未检测到“OK”或“ERROR”，写入线程应直接发送下一条AT命令（即读取线程无响应时，写入线程仍需每3秒发送一条命令，因调制解调器AT命令接口常不稳定）。

目前我通过mpsc::channel实现了一个临时方案：

初始化代码 #

let (sender, receiver) = channel::<()>();

调制解调器AT命令读取线程代码 #

if line.starts_with("OK") || line.contains("ERROR") {
    debug!("Sending go-ahead to writing_thread.");
    sender.send(()).unwrap();
}

调制解调器AT命令写入线程代码 #

/* This receive is just a way of blocking until the modem is ready. */
match receiver.recv_timeout(Duration::from_secs(3)) {
    Ok(_) => {
        debug!("Received go-ahead from reading thread.");
        /*
         * Empty the channel, in case the modem was too effusive.  We don't want
         * to "bank" earlier OK/ERRORs to allow multiple AT commands to be sent in
         * quick succession.
         */
        while let Ok(_) = receiver.try_recv() {}
    }
    Err(err) => match err {
        RecvTimeoutError::Timeout => {
            debug!("Timed-out waiting for go-ahead from reading thread.");
        }
        RecvTimeoutError::Disconnected => break 'outer
    },
}

我未找到Mutex::lock_with_timeout()方法，请问如何使用Mutex<()>或类似工具正确实现该需求？

标准库的Mutex本身没有超时锁方法，但可以搭配Condvar（条件变量）实现带超时的同步逻辑，完全匹配你的需求：

核心思路 #

用Mutex<bool>标记是否收到读取线程的“OK/ERROR”信号，Condvar负责通知写入线程状态变化。写入线程等待条件变量，超时则直接执行下一步；读取线程检测到目标响应时，更新互斥锁内的状态并触发通知。

具体实现代码 #

初始化 #

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

// 封装同步状态：是否已收到就绪信号
struct ReadyState {
    is_ready: bool,
}

let state = Arc::new((
    Mutex::new(ReadyState { is_ready: false }),
    Condvar::new(),
));
let state_clone = Arc::clone(&state);

读取线程逻辑 #

// 读取线程中处理响应的代码段
let (lock, cvar) = &*state_clone;
if line.starts_with("OK") || line.contains("ERROR") {
    debug!("Sending go-ahead to writing_thread.");
    let mut ready_state = lock.lock().unwrap();
    ready_state.is_ready = true;
    // 通知等待的写入线程
    cvar.notify_one();
}

写入线程逻辑 #

// 写入线程的核心循环
let (lock, cvar) = &*state;
loop {
    let mut ready_state = lock.lock().unwrap();
    // 等待就绪信号，或超时3秒
    let result = cvar.wait_timeout(ready_state, Duration::from_secs(3)).unwrap();
    ready_state = result.0;
    let timed_out = result.1.timed_out();

    if ready_state.is_ready {
        debug!("Received go-ahead from reading thread.");
        // 重置状态，避免复用旧信号
        ready_state.is_ready = false;
    } else if timed_out {
        debug!("Timed-out waiting for go-ahead from reading thread.");
    }

    // 执行发送AT命令的逻辑
    // send_at_command(...);

    // 可自行添加循环退出条件，如连接断开等
}

方案说明 #

1. 避免信号堆积：每次处理完成后重置is_ready为false，和你原方案清空channel的逻辑一致，防止读取线程多次触发导致写入线程连续发送命令。
2. 超时与通知兼容：wait_timeout会明确返回是否超时，无论哪种情况，写入线程都能继续执行发送逻辑。
3. 方案对比：你现有的mpsc方案已经能很好满足需求，Condvar+Mutex属于更传统的同步模式，两种方案各有优劣——mpsc实现更简洁，Condvar+Mutex则更直观地体现同步状态。

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