Rust消息总线库bus的使用：高效实现跨线程通信与事件驱动的轻量级解决方案

Bus 提供了一个无锁、有界、单生产者、多消费者的广播通道。

注意：当前实现中bus有时会忙等待，这可能导致CPU使用率增加

特性

• 使用环形缓冲区和原子指令实现无锁单生产者、多消费者通道
• 接口与std::sync::mpsc通道类似，但支持多个消费者
• 广播模式：每个发送都会传递给每个消费者
• 线程安全设计

安装

在Cargo.toml中添加依赖：

bus = "2.4.1"

基本用法示例

use bus::Bus;

fn main() {
    // 创建一个容量为10的消息总线
    let mut bus = Bus::new(10);
    
    // 添加两个接收者
    let mut rx1 = bus.add_rx();
    let mut rx2 = bus.add_rx();
    
    // 发送消息
    bus.broadcast("Hello");
    bus.broadcast("world");
    
    // 接收消息
    assert_eq!(rx1.recv(), Ok("Hello"));
    assert_eq!(rx1.recv(), Ok("world"));
    
    assert_eq!(rx2.recv(), Ok("Hello")); 
    assert_eq!(rx2.recv(), Ok("world"));
}

完整示例：跨线程事件总线

use bus::Bus;
use std::thread;
use std::time::Duration;

fn main() {
    // 创建事件总线，容量为5
    let mut event_bus = Bus::new(5);
    
    // 创建3个消费者线程
    for i in 0..3 {
        let mut rx = event_bus.add_rx();
        thread::spawn(move || {
            loop {
                match rx.recv() {
                    Ok(msg) => println!("消费者{}收到: {}", i, msg),
                    Err(_) => break, // 总线关闭时退出
                }
            }
        });
    }
    
    // 生产者线程
    thread::spawn(move || {
        for i in 1..=5 {
            event_bus.broadcast(format!("消息{}", i));
            thread::sleep(Duration::from_millis(500));
        }
        // 总线会在event_bus离开作用域时自动关闭
    });
    
    // 等待所有消息处理完成
    thread::sleep(Duration::from_secs(3));
}

高级用法

use bus::Bus;

fn main() {
    let mut bus = Bus::new(3);
    
    // 添加接收者
    let mut rx1 = bus.add_rx();
    let mut rx2 = bus.add_rx();
    
    // 发送不同类型的数据
    bus.broadcast(42);
    bus.broadcast("hello");
    bus.broadcast(vec![1, 2, 3]);
    
    // 使用try_recv非阻塞接收
    assert_eq!(rx1.try_recv(), Ok(42));
    assert_eq!(rx1.try_recv(), Ok("hello"));
    assert_eq!(rx1.try_recv(), Ok(vec![1, 2, 3]));
    
    // 同样适用于rx2
    assert_eq!(rx2.try_recv(), Ok(42));
    assert_eq!(rx2.try_recv(), Ok("hello"));
    assert_eq!(rx2.try_recv(), Ok(vec![1, 2, 3]));
    
    // 总线已满时发送会阻塞
    // 可以使用try_broadcast进行非阻塞发送
    assert!(bus.try_broadcast(99).is_ok());
    assert!(bus.try_broadcast(100).is_ok());
    assert!(bus.try_broadcast(101).is_err()); // 总线已满
}

注意事项

1. 这是一个单生产者、多消费者的广播通道
2. 所有消息都会被所有消费者接收
3. 通道有固定容量，满时会阻塞发送
4. 当前实现可能存在忙等待问题

许可证

可选择 MIT 或 Apache-2.0 许可证

完整示例demo

下面是一个更完整的示例，展示如何使用bus库实现一个简单的事件驱动系统：

use bus::Bus;
use std::thread;
use std::time::Duration;

// 定义事件枚举
#[derive(Debug, Clone)]
enum Event {
    Start,
    Data(String),
    Stop,
}

fn main() {
    // 创建事件总线，容量为10
    let mut bus = Bus::new(10);
    
    // 创建3个消费者线程
    for i in 0..3 {
        let mut rx = bus.add_rx();
        thread::spawn(move || {
            loop {
                match rx.recv() {
                    Ok(event) => {
                        match event {
                            Event::Start => println!("消费者{}: 系统启动", i),
                            Event::Data(msg) => println!("消费者{}: 处理数据 - {}", i, msg),
                            Event::Stop => {
                                println!("消费者{}: 系统停止", i);
                                break;
                            }
                        }
                    },
                    Err(_) => break, // 总线关闭时退出
                }
            }
        });
    }
    
    // 生产者线程
    thread::spawn(move || {
        // 发送启动事件
        bus.broadcast(Event::Start);
        
        // 发送一些数据事件
        for i in 1..=5 {
            bus.broadcast(Event::Data(format!("消息{}", i)));
            thread::sleep(Duration::from_millis(300));
        }
        
        // 发送停止事件
        bus.broadcast(Event::Stop);
        
        // 总线会在bus离开作用域时自动关闭
    });
    
    // 等待所有消息处理完成
    thread::sleep(Duration::from_secs(3));
}

这个完整示例展示了：

1. 定义自定义事件类型
2. 多个消费者线程处理不同类型的事件
3. 生产者线程发送不同类型的事件
4. 优雅地处理系统启动和停止
5. 使用枚举来区分不同的事件类型