如何实现最大长度为1的mpsc::channel单元素通道？

针对你的需求——生产者发送消息，消费者只关注最新消息、自动丢弃旧消息，同时实现所有权转移且尽可能缩短消费者阻塞时间——可以通过以下两种方案实现：

方案一：无锁实现（基于AtomicPtr） #

利用原子指针实现无锁操作，完全避免锁竞争的阻塞开销，适合对性能要求极高的场景。

use std::sync::Arc;
use std::sync::atomic::{AtomicPtr, Ordering};
use std::ptr;

// 单元素通道核心结构
struct SingleElementChannel<T> {
    ptr: AtomicPtr<T>,
}

impl<T> SingleElementChannel<T> {
    // 创建生产者和消费者对
    fn new() -> (Sender<T>, Receiver<T>) {
        let channel = Arc::new(SingleElementChannel {
            ptr: AtomicPtr::new(ptr::null_mut()),
        });
        (
            Sender { channel: channel.clone() },
            Receiver { channel, last_msg: None }
        )
    }
}

// 生产者端
struct Sender<T> {
    channel: Arc<SingleElementChannel<T>>,
}

impl<T> Sender<T> {
    // 发送消息：自动替换并丢弃旧消息
    fn send(&self, msg: T) {
        let new_ptr = Box::into_raw(Box::new(msg));
        // 原子交换旧指针，获取所有权后释放旧消息
        let old_ptr = self.channel.ptr.swap(new_ptr, Ordering::Release);
        if !old_ptr.is_null() {
            unsafe { Box::from_raw(old_ptr); }
        }
    }
}

// 消费者端
struct Receiver<T> {
    channel: Arc<SingleElementChannel<T>>,
    last_msg: Option<T>,
}

impl<T> Receiver<T> {
    // 获取最新消息：无新消息则返回上次的结果
    fn get_latest(&mut self) -> &T {
        let current_ptr = self.channel.ptr.load(Ordering::Acquire);
        if !current_ptr.is_null() {
            // 原子交换将通道置空，转移消息所有权
            let msg_ptr = self.channel.ptr.swap(ptr::null_mut(), Ordering::Acquire);
            if !msg_ptr.is_null() {
                let new_msg = unsafe { *Box::from_raw(msg_ptr) };
                self.last_msg = Some(new_msg);
            }
        }
        // 注意：若从未收到过消息会panic，可根据需求改为返回Option<&T>或提供默认值
        self.last_msg.as_ref().unwrap()
    }
}

// 通道销毁时清理剩余消息，避免内存泄漏
impl<T> Drop for SingleElementChannel<T> {
    fn drop(&mut self) {
        let ptr = self.ptr.load(Ordering::Relaxed);
        if !ptr.is_null() {
            unsafe { Box::from_raw(ptr); }
        }
    }
}

实现说明： #

• 生产者发送消息时，将新消息包装为Box并转为原始指针，通过原子交换替换通道内的旧指针，旧消息会被自动释放，不会堆积无用数据。
• 消费者获取消息时，通过原子操作获取最新消息的所有权，更新本地缓存的last_msg，无新消息则直接返回缓存值。
• 使用unsafe代码处理原始指针，但通过严格的空指针检查和Drop实现保证内存安全。

方案二：Mutex轻量实现 #

如果对极致性能要求不高，使用Mutex的实现更简单安全，锁的持有时间极短（仅替换/取出消息的瞬间），实际阻塞可以忽略。

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

struct SingleElementChannel<T> {
    msg: Mutex<Option<Box<T>>>,
}

impl<T> SingleElementChannel<T> {
    fn new() -> (Sender<T>, Receiver<T>) {
        let channel = Arc::new(SingleElementChannel {
            msg: Mutex::new(None),
        });
        (
            Sender { channel: channel.clone() },
            Receiver { channel, last_msg: None }
        )
    }
}

struct Sender<T> {
    channel: Arc<SingleElementChannel<T>>,
}

impl<T> Sender<T> {
    fn send(&self, msg: T) {
        let mut guard = self.channel.msg.lock().unwrap();
        // 替换旧消息，旧消息自动被drop
        *guard = Some(Box::new(msg));
    }
}

struct Receiver<T> {
    channel: Arc<SingleElementChannel<T>>,
    last_msg: Option<T>,
}

impl<T> Receiver<T> {
    fn get_latest(&mut self) -> &T {
        let mut guard = self.channel.msg.lock().unwrap();
        // 取出最新消息（若存在），转移所有权到本地缓存
        if let Some(new_msg_box) = guard.take() {
            self.last_msg = Some(*new_msg_box);
        }
        // 同样，未收到过消息时会panic，可按需调整
        self.last_msg.as_ref().unwrap()
    }
}

实现说明： #

• 用Mutex保护一个Option<Box<T>>，生产者发送时锁定Mutex替换消息，消费者获取时锁定Mutex取出最新消息。
• 无需unsafe代码，内存安全由标准库保证，实现成本更低。

方案对比 #

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