publishing/subscribing 在 Rust 中的良好服务器架构

Question

我正在尝试用 Rust 实现一个服务器，我想知道我是否遗漏了一些东西来降低它的复杂性。我的项目的简化是我想让不同的客户端连接到服务器，每个客户端实时接收其他客户端发布的内容。

我想在异步运行时 (tokio atm) 中执行此操作。每个client都有自己的stream，我把stream分成stream和sink:

let (split_sink, split_stream) = stream.split();

我只想以最简单的方式将每个 split_stream 连接到每个其他 split_sink。

我已经使用 Arc 和 Mutex 解决了这个问题。

由于需要从处理新客户端连接的任务和处理该发布者已发布消息的任务访问每个发布者中的订阅者列表，我将它们包装在 Arc 和 Mutex 中。
由于每个订阅者需要从多个发布者访问，我将它们包装在 Arc 和 Mutex 中。

这给我留下了这样的东西：

pub struct Publisher {
    subscribers: Arc<Mutex<Vec<Arc<Mutex<Subscriber>>>>>,
    ...
}

impl Publisher {
    ...
    pub async fn add(&self, subscriber: Arc<Mutex<Subscriber>>) {
        self.subscribers.lock().await.push(subscriber);
    }
    ...
    async fn forward_message(
        message: Message,
        subscribers: &Arc<Mutex<Vec<Arc<Mutex<Subscriber>>>>>,
    ) {
        for subscriber in subscribers.lock().await.iter() {
            let mut subscriber = subscriber.lock().await;
            subscriber.split_sink.send(message.clone()).await.unwrap();
        }
    }
}

我对 Rust 很陌生，这对我来说看起来很糟糕，过于复杂。我可以做任何简化吗？诸如更好的并发数据结构或不同的架构之类的东西？提前谢谢你。

Answer 1

tokio::sync::broadcast 似乎很符合你的要求。

如果你想自己实现，线程间通信的选项包括：

std::sync：原子、同步对象等
tokio::sync：同步对象的异步兼容版本，各种通道等
futures_core::stream/tokio_stream：像 Iterator 但异步

publishing/subscribing 在 Rust 中的良好服务器架构

Good server architecture for publishing/subscribing in Rust

concurrency

ownership

publish-subscribe

rust

async-await