Java最简线程安全消息传递方案的缺陷及相关问题问询

一、并发访问问题的修复方案 #

你的原始代码存在竞态条件和内存可见性问题：多线程访问state和message时无同步保障，且receive方法错误地将状态设为SENT（会导致生产者永久阻塞）。要解决这些问题，需用Java同步机制保证原子性和可见性，同时用wait/notify替代低效的Thread.sleep忙等。

修复后的代码如下：

class MessagePasser<T> {
    enum MessagePasserState { SENT, RECEIVED }

    private MessagePasserState state = MessagePasserState.RECEIVED;
    private T message;

    public synchronized void send(T message) throws InterruptedException {
        // 等待消费者取走上一条消息
        while (state == MessagePasserState.SENT) {
            wait();
        }
        this.message = message;
        state = MessagePasserState.SENT;
        // 通知消费者有新消息
        notify();
    }

    public synchronized T receive() throws InterruptedException {
        // 等待生产者发送消息
        while (state == MessagePasserState.RECEIVED) {
            wait();
        }
        T message = this.message;
        // 修正状态为已接收，允许生产者发下一条
        state = MessagePasserState.RECEIVED;
        // 通知生产者可以继续发送
        notify();
        return message;
    }
}

• synchronized保证同一时间只有一个线程进入方法，避免竞态条件；
• wait/notify让线程在无消息时进入阻塞状态，不会空耗CPU；
• 修正了receive方法的状态切换错误，这是原始代码的核心bug。

二、原始方案的其他缺陷 #

除了并发问题，原始代码还有以下明显问题：

• 低效的忙等：Thread.sleep(50)是定时轮询，无论消息状态是否变化都会唤醒线程，浪费CPU资源；
• 内存可见性缺失：state和message未用volatile修饰，一个线程的修改对其他线程可能不可见，导致线程无限循环等待；
• 单消息瓶颈：同一时间只能传递一条消息，生产者必须等消费者取走才能发下一条，吞吐量极低；
• 异常处理缺失：Thread.sleep会抛出InterruptedException，但代码未处理，导致线程中断逻辑无法正常工作。

三、添加缓存机制的难度 #

如果要支持多条消息缓存（消息队列），改造起来不算复杂：

1. 将单message字段替换为队列（如LinkedList<T>）；
2. 修改send逻辑：队列未满时直接入队，满则等待；修改receive逻辑：队列非空时直接出队，空则等待；
3. 可用Condition分别处理队列满/空场景（比wait/notify更灵活）。

其实Java标准库已提供现成的线程安全队列（如LinkedBlockingQueue、ArrayBlockingQueue），直接使用这些工具类可快速实现带缓存的消息传递，无需重复造轮子。

四、其他语言的实现方式 #

C语言 #

用POSIX线程库的互斥锁和条件变量实现，核心逻辑与Java类似：

#include <pthread.h>
#include <stdlib.h>

typedef enum { SENT, RECEIVED } MessagePasserState;

typedef struct {
    MessagePasserState state;
    void* message;
    pthread_mutex_t mutex;
    pthread_cond_t cond;
} MessagePasser;

void message_passer_init(MessagePasser* passer) {
    passer->state = RECEIVED;
    passer->message = NULL;
    pthread_mutex_init(&passer->mutex, NULL);
    pthread_cond_init(&passer->cond, NULL);
}

void send(MessagePasser* passer, void* message) {
    pthread_mutex_lock(&passer->mutex);
    while (passer->state == SENT) {
        pthread_cond_wait(&passer->cond, &passer->mutex);
    }
    passer->message = message;
    passer->state = SENT;
    pthread_cond_signal(&passer->cond);
    pthread_mutex_unlock(&passer->mutex);
}

void* receive(MessagePasser* passer) {
    pthread_mutex_lock(&passer->mutex);
    while (passer->state == RECEIVED) {
        pthread_cond_wait(&passer->cond, &passer->mutex);
    }
    void* message = passer->message;
    passer->state = RECEIVED;
    pthread_cond_signal(&passer->cond);
    pthread_mutex_unlock(&passer->mutex);
    return message;
}

void message_passer_destroy(MessagePasser* passer) {
    pthread_mutex_destroy(&passer->mutex);
    pthread_cond_destroy(&passer->cond);
}

C++语言 #

用标准库的std::mutex和std::condition_variable实现，支持泛型：

#include <mutex>
#include <condition_variable>

enum class MessagePasserState { SENT, RECEIVED };

template<typename T>
class MessagePasser {
private:
    MessagePasserState state = MessagePasserState::RECEIVED;
    T message;
    std::mutex mtx;
    std::condition_variable cv;

public:
    void send(const T& msg) {
        std::unique_lock<std::mutex> lock(mtx);
        // 等待消息被取走
        cv.wait(lock, [this](){ return state == MessagePasserState::RECEIVED; });
        message = msg;
        state = MessagePasserState::SENT;
        cv.notify_one();
    }

    T receive() {
        std::unique_lock<std::mutex> lock(mtx);
        // 等待消息被发送
        cv.wait(lock, [this](){ return state == MessagePasserState::SENT; });
        T msg = message;
        state = MessagePasserState::RECEIVED;
        cv.notify_one();
        return msg;
    }
};

Rust语言 #

Rust的所有权系统和线程安全机制能让实现更简洁安全，无需手动管理锁的细节。

基于mpsc通道的实现（天然支持多生产者单消费者，带缓存）

Rust标准库的mpsc通道是常用的消息传递方式，默认支持无界缓存，也可创建有界缓存：

use std::sync::mpsc;
use std::thread;

fn main() {
    // 创建通道，返回生产者(sender)和消费者(receiver)
    let (sender, receiver) = mpsc::channel();

    // 生产者线程
    thread::spawn(move || {
        if sender.send("Hello from Rust producer").is_ok() {
            println!("Producer sent message");
        }
    });

    // 消费者线程
    thread::spawn(move || {
        if let Ok(msg) = receiver.recv() {
            println!("Consumer received: {}", msg);
        }
    }).join().unwrap();
}

单消息所有权传递实现

如果要贴合原始方案“生产者阻塞直到消费者取走”的逻辑，可用Arc<Mutex<Option<T>>>结合Condvar，利用Rust的所有权转移保证消息只能被取走一次：

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

struct MessagePasser<T> {
    data: Mutex<Option<T>>,
    cond: Condvar,
}

impl<T> MessagePasser<T> {
    fn new() -> Self {
        MessagePasser {
            data: Mutex::new(None),
            cond: Condvar::new(),
        }
    }

    fn send(&self, msg: T) {
        let mut data = self.data.lock().unwrap();
        // 等待消息被取走（data为None）
        while data.is_some() {
            data = self.cond.wait(data).unwrap();
        }
        *data = Some(msg);
        // 通知消费者有新消息
        self.cond.notify_one();
    }

    fn receive(&self) -> T {
        let mut data = self.data.lock().unwrap();
        // 等待消息被发送（data为Some）
        while data.is_none() {
            data = self.cond.wait(data).unwrap();
        }
        // 转移消息所有权，确保只能取走一次
        data.take().unwrap()
    }
}

fn main() {
    let passer = Arc::new(MessagePasser::new());
    let passer_clone = passer.clone();

    // 生产者线程
    thread::spawn(move || {
        passer.send("Hello from single-message passer".to_string());
        println!("Producer finished sending");
    });

    // 消费者线程
    let handle = thread::spawn(move || {
        let msg = passer_clone.receive();
        println!("Consumer got: {}", msg);
    });

    handle.join().unwrap();
}

Rust编译器会在编译阶段检查线程安全，避免数据竞争，所有权转移机制也能避免重复访问消息的问题，逻辑更可靠。

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