如何在Rust发布订阅系统中用HashMap存储对象引用？

问题背景 #

要实现一个迷你Kafka风格的事件处理系统：

• 用HashMap存储消费者，键为对象名称，值为Consumer的共享不可变引用
• Event发送到Topic后推送给所有订阅的Consumer
• 实现Topic::subscribe和unsubscribe时触发编译器错误：提示将Consumer移动到了Topic中，但仅想存储/移除共享引用，尝试Rc智能指针未解决

核心原因 #

原代码的实现违反了Rust所有权规则：

• 直接存储&Consumer会有生命周期限制，无法长期持有
• 尝试移动Consumer实例会转移所有权，导致原对象无法再使用
• 错误使用Rc：从共享引用中移动对象创建Rc，触发E0507移动错误

解决方案 #

使用Rc<Consumer>实现共享所有权，让Topic和外部代码同时持有Consumer的引用（仅增加引用计数，不移动对象），具体修改如下：

1. 导入Rc智能指针 #

在代码顶部添加：

use std::rc::Rc;

2. 修改Topic的订阅者存储类型 #

将subscribers: HashMap<String, Consumer>改为：

subscribers: HashMap<String, Rc<Consumer>>,

3. 修正subscribe方法 #

接收Rc<Consumer>作为参数，直接插入HashMap（Rc的clone仅增加引用计数，无性能开销）：

fn subscribe(&mut self, consumer: Rc<Consumer>) {
    self.subscribers.entry(consumer.name.clone()).or_insert(consumer);
}

4. 修正unsubscribe方法 #

无需传递整个Consumer实例，仅通过名称移除即可：

fn unsubscribe(&mut self, consumer_name: &str) {
    self.subscribers.remove(consumer_name);
}

5. 修复Consume trait的实例方法 #

原trait的consume是关联函数，改为实例方法才能针对每个Consumer处理事件：

pub trait Consume {
    fn consume(&self, event: Event) {
        // 默认实现
    }
}

impl Consume for Consumer {
    fn consume(&self, event: Event) {
        println!("{} [{}]: {}", self.name, event.etype, event.msg);
    }
}

完整修改后的代码 #

use std::collections::HashMap;
use std::rc::Rc;

// 事件结构体
pub struct Event {
    etype: String,
    msg: String,
}

// 消费者 trait
pub trait Consume {
    fn consume(&self, event: Event) {
        
    }
}

#[derive(PartialEq, Clone)]
struct Consumer {
    name: String,
}

impl Consume for Consumer {
    fn consume(&self, event: Event) {
        println!("{} [{}]: {}", self.name, event.etype, event.msg);
    }
}

pub struct Topic {
    name: String,
    subscribers: HashMap<String, Rc<Consumer>>,
    events: Vec<Event>,
}

impl Topic {
    fn new(tname: String) -> Topic {
        Topic {
            name: tname,
            subscribers: HashMap::new(),
            events: Vec::new(),
        }
    }

    fn subscribe(&mut self, consumer: Rc<Consumer>) {
        self.subscribers.entry(consumer.name.clone()).or_insert(consumer);
    }

    fn unsubscribe(&mut self, consumer_name: &str) {
        self.subscribers.remove(consumer_name);
    }

    // 新增：推送事件给所有订阅者
    fn publish(&self, event: Event) {
        for consumer in self.subscribers.values() {
            consumer.consume(event.clone());
        }
    }
}

修正后的测试代码 #

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_create_event() {
        let type_name = "Test";
        let message = "Hello, World!";

        let event = Event {
            etype: type_name.to_string(),
            msg: message.to_string()
        };

        assert_eq!(event.etype, type_name);
        assert_eq!(event.msg, message);
    }

    #[test]
    fn test_topic_create() {
        let topic_name = "Test";
        let topic = Topic::new(topic_name.to_string());

        assert_eq!(topic.name, topic_name);
    }

    #[test]
    fn test_topic_subscribe() {
        let topic_name = "Topic";
        let mut topic = Topic::new(topic_name.to_string());

        let consumer_name = "Consumer";
        let cons = Rc::new(Consumer { name: consumer_name.to_string() });

        topic.subscribe(cons.clone());

        let test_subscriber = topic.subscribers[consumer_name].name.clone();
        assert_eq!(test_subscriber, consumer_name);
        // 验证原Rc仍可正常使用
        assert_eq!(cons.name, consumer_name);
    }

    #[test]
    fn test_topic_unsubscribe() {
        let topic_name = "Topic";
        let mut topic = Topic::new(topic_name.to_string());

        let consumer_name = "Consumer";
        let cons = Rc::new(Consumer { name: consumer_name.to_string() });

        topic.subscribe(cons);
        topic.unsubscribe(consumer_name);

        assert!(topic.subscribers.get(consumer_name).is_none());
    }
}

关键说明 #

• Rc<Consumer>允许多个所有者共享同一个Consumer实例，每次clone仅增加引用计数，无额外性能消耗
• Topic存储Rc后，外部代码仍可持有Consumer的Rc引用，不会失去对对象的控制权
• 移除订阅时仅需通过名称操作，更贴合Kafka的设计逻辑

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