Rust中的Event Bus与观察者模式：实现与示例

Rust中的Event Bus与观察者模式：实现与示例

Event Bus（事件总线）是一种实现发布-订阅（publish-subscribe）原则的架构模式。在Rust中，该模式对于构建松耦合系统特别有用，组件之间通过事件进行通信，而无需直接了解彼此。在本文中，我们将探讨如何使用观察者模式创建自己的Event Bus，并介绍Rust生态系统中的现成解决方案。

观察者模式：基础

观察者模式定义了对象之间的一对多关系。当一个对象（发布者）改变其状态时，所有依赖对象（订阅者）都会自动收到通知。在Rust中，这通常通过trait和动态分发来实现。

Rust中观察者模式的基本实现

use std::collections::HashMap;

// 观察者的traittrait Observer {    fn update(&self, event: &str);}

// 发布者的结构体struct EventManager {    observers: Vec<Box<dyn Observer>>,}

impl EventManager {    fn new() -> Self {        EventManager {            observers: Vec::new(),        }    }

    fn subscribe(&mut self, observer: Box<dyn Observer>) {        self.observers.push(observer);    }

    fn notify(&self, event: &str) {        for observer in &self.observers {            observer.update(event);        }    }}

// 具体的观察者struct Logger;

impl Observer for Logger {    fn update(&self, event: &str) {        println!("[Logger] 收到事件: {}", event);    }}

fn main() {    let mut manager = EventManager::new();    manager.subscribe(Box::new(Logger));    manager.notify("user.login");}

Event Bus：观察者模式的扩展版本

Event Bus是一个集中式总线，根据事件类型进行路由。与简单的观察者模式不同，它支持不同类型的事件和订阅过滤。

带事件类型的Event Bus实现

use std::any::Any;use std::collections::HashMap;use std::sync::{Arc, Mutex};

// 事件类型#[derive(Clone, Debug)]enum EventType {    UserLogin,    UserLogout,    DataUpdate,}

// 事件结构体#[derive(Clone, Debug)]struct Event {    event_type: EventType,    payload: String,}

// 事件处理器type Handler = Arc<dyn Fn(Event) + Send + Sync>;

// 事件总线struct EventBus {    handlers: HashMap<EventType, Vec<Handler>>,}

impl EventBus {    fn new() -> Self {        EventBus {            handlers: HashMap::new(),        }    }

    fn subscribe<F>(&mut self, event_type: EventType, handler: F)    where        F: Fn(Event) + 'static + Send + Sync,    {        self.handlers            .entry(event_type)            .or_insert_with(Vec::new)            .push(Arc::new(handler));    }

    fn publish(&self, event: Event) {        if let Some(handlers) = self.handlers.get(&event.event_type) {            for handler in handlers {                handler(event.clone());            }        }    }}

fn main() {    let mut bus = EventBus::new();

    bus.subscribe(EventType::UserLogin, |event| {        println!("用户登录: {}", event.payload);    });

    bus.subscribe(EventType::UserLogin, |event| {        // 另一个处理器        println!("登录已记录: {}", event.payload);    });

    bus.publish(Event {        event_type: EventType::UserLogin,        payload: "user123".to_string(),    });}

使用Tokio的异步Event Bus

在实际应用中，通常需要异步事件处理。为此，tokio库及其broadcast通道非常合适。

使用tokio::sync::broadcast

use tokio::sync::broadcast;

#[tokio::main]async fn m