Rust异步BehaviorSubject实现求助：状态共享与流优化问题

需求说明 #

需要实现一个BehaviorSubject<T>结构体，作为异步上下文的可变状态共享容器，满足：

• 支持克隆，可跨线程/任务传递（需实现Clone + Send + Sync，用于Axum应用状态）
• 可转换为流：初始发送当前值，后续推送所有状态更新（仅保留最新值，忽略中间滞后事件）
• 支持获取/设置内部值：状态变更时，所有克隆实例同步更新，所有订阅流收到通知

现有实现的Bug #

基于Tokio broadcast channel和Mutex的现有实现存在两个核心问题：

1. 慢接收者阻塞其他操作：into_stream方法中先加锁获取初始值，若有慢接收者在此阶段阻塞，会导致所有后续的get_value/set_value操作等待锁释放，拖慢整体性能。
2. 生命周期清理异常：当所有BehaviorSubject实例被丢弃（仅保留流订阅）时，流不会自动终止——因为broadcast的Sender被所有BehaviorSubject克隆持有，最后一个实例转为流后，Sender仍存在，通道不会关闭。

改进实现方案 #

解决锁阻塞问题 #

调整初始值获取逻辑，缩短锁的持有时间：仅在克隆初始值时持有锁，避免长时间占用锁导致其他操作阻塞。同时，将内部状态封装到Arc包裹的结构体中，优化克隆成本。

修复生命周期清理 #

利用Arc的自动计数机制：当所有BehaviorSubject实例被丢弃时，内部的Sender会被自动销毁，broadcast通道关闭，所有订阅流会收到Closed错误并终止。

改进后代码 #

use futures::Stream;
use std::sync::Arc;
use tokio::sync::{
    broadcast::{channel, error::RecvError, Sender},
    Mutex,
};

#[derive(Clone)]
pub struct BehaviorSubject<T> {
    inner: Arc<BehaviorSubjectInner<T>>,
}

struct BehaviorSubjectInner<T> {
    value: Mutex<T>,
    sender: Sender<T>,
}

impl<T: Clone> BehaviorSubject<T> {
    pub fn new(initial_value: T) -> Self {
        // 缓冲区设为1，只保留最新更新
        let (sender, _) = channel(1);
        BehaviorSubject {
            inner: Arc::new(BehaviorSubjectInner {
                value: Mutex::new(initial_value),
                sender,
            }),
        }
    }

    pub async fn get_value(&self) -> T {
        self.inner.value.lock().await.clone()
    }

    pub async fn set_value(&self, new_value: T) {
        let mut locked_value = self.inner.value.lock().await;
        let cloned_value = new_value.clone();
        *locked_value = new_value;
        // 忽略无订阅者的发送错误
        let _ = self.inner.sender.send(cloned_value);
    }

    pub fn into_stream(self) -> impl Stream<Item = T> {
        async_stream::stream! {
            // 快速获取初始值，锁仅用于克隆
            let initial = self.inner.value.lock().await.clone();
            yield initial;

            let mut rx = self.inner.sender.subscribe();
            loop {
                match rx.recv().await {
                    Ok(value) => yield value,
                    Err(RecvError::Lagged(_)) => continue, // 跳过滞后的旧值
                    Err(RecvError::Closed) => break,       // 通道关闭时终止流
                }
            }
        }
    }
}

// 自动推导Send/Sync，只要T满足条件
unsafe impl<T: Send + Sync> Send for BehaviorSubject<T> {}
unsafe impl<T: Send + Sync> Sync for BehaviorSubject<T> {}

进一步优化：无锁状态访问 #

如果T支持原子操作，可使用atomic crate的AtomicCell<T>替代Mutex<T>，彻底消除锁竞争：

use atomic::AtomicCell;
// 其他导入保持不变

struct BehaviorSubjectInner<T> {
    value: AtomicCell<T>,
    sender: Sender<T>,
}

impl<T: Clone + Default> BehaviorSubject<T> {
    pub async fn get_value(&self) -> T {
        self.inner.value.load().clone()
    }

    pub async fn set_value(&self, new_value: T) {
        let cloned_value = new_value.clone();
        self.inner.value.store(new_value);
        let _ = self.inner.sender.send(cloned_value);
    }
}

替代库推荐 #

若不想手动实现，可直接使用成熟的第三方库：

• rxrust：Rust原生的响应式编程库，内置BehaviorSubject，完全匹配需求
• async-broadcast：比Tokio broadcast更灵活的异步广播实现，支持自定义缓冲区和清理策略
• tokio-util：提供BroadcastStream，可结合初始值快速封装类BehaviorSubject的流

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