在阅读本小节前请读者先自己尝试回答这个问题：Subscriber 类真的有必要实现吗？

在我们上面的代码中，Subscriber 类 实现了几乎唯一一个有用的方法：update，它的作用却是给 Publisher 类的 notifyAll 方法进行调用。

从现实生活中给一个解释：
notifyAll 是消息的发布这发布消息的工具，update 是订阅用户接受到的新的定制化消息，比如同样是订阅了售房信息，但是由于不同类别的购房者订阅时所选定的楼层、大小等参数不一样，则这些不同订阅者接收到的发布结果不一样——也就表面原始的消息需要为不同的订阅者做一些 定制化 处理。在之前的代码实现中，这个消息的定制化工作就是使用 Subscriber 类的 update 方法实现的。

很显然，上表面的代码要真正实现定制化，往往不仅是参数值的不同，可能对参数的处理也不一样。因此仅仅依赖参数data是不合理的。因此我们大概是需要写多个仅仅 update 方法的实现不同的 Subscriber 类——这不太好。略好一点的办法是，让 update 方法接受的不是单纯的数据 data，而是一个 回调函数 传入 update 方法中。

先不着急修改我们的代码。对于发布者来说，似乎可以提供 更加周到 的服务——直接登记好订阅着的定制化需求处理方式，使用订阅者要求的处理方式处理好定制的消息后，直接告诉订阅者。——因此 update 这个接受表示用户定制化需求处理方式的方法可以直接合并到 发布者那边。

于是 Subscriber 类 就不需要了，现在我们只需要更新一下我们的 Publisher。更新的思路是这样的：

• 添加订阅者时（Publisher.addSubscriber）不仅需要记录订阅者名字，还要记录一个对应的响应函数用以消息发布后给订阅者提供定制化服务。

从 Publisher 看，需要登记的内容又多了一些。不过好在 订阅者名称（认为是唯一标识符）和 与之对于的服务（回调函数），是对应的关系，既可以一对一，也可以一对多（表示这个订阅者需要多个定制化服务）。因此我们将 Publisher 的 “记录本”改成下面的类型：

HashMap<String, Vec<EventCallback>>

这个映射（Map）的 key 就表示 订阅者名称，而value 部分是一组函数，表示该订阅者需要的各种服务。

另外，到了这里，对于 Publisher来说，添加订阅者就转化为了 为订阅者订阅各种定制化服务 。同时反过来看，对于某个具体的订阅者 Subscriber，一旦它的服务定制数组 （Function[]）为空数组，表明他已经没有任何订阅，也不再需要接收发布者的任何消息了。

因此先前我们使用的方法名addSubscriber 不适用了，从含以上换成 addListener 似乎更加合适。

为什么呢? 我们接下来对此做进一步说明。

一直以来，我们聚焦点都在于 发布者 和 订阅者，而忽略了 引起发布者发布的事件。 这个方法接受两个参数，一个是用户名，一个是为用户新增的回调。同时必须指出的是，这个 回调 往往是需要再其调用时接受一些数据的，比如由发布者发布的某些原始数据，他们就像是时时刻刻地 监听着、守候着 发布者 发布一个事件，一旦这个 事件/消息 被发布，就 完成消息发布后为 订阅者 所提供地服务。

换一下思路，我们接着把聚焦点转移到 事件 上来。

其实从现实中看，同一个事件发生，可能意味着可能需 要干很多事，既可以 服务更多的订阅者，也可以干其它任何的工作——我们一味地想着在发布者处登记订阅者的id然后完成订阅者的需求，那么 没有区别为何事件而需要去发布这些消息！

更好的做法是 不再记录订阅者，而是记录为什么要发布消息给订阅者——也就是记录 事件。这样我们就可以在同一个事件发生的时候，通过一系列的属于该事件函数（可能一个或多个回调函数服务于同一个订阅者），完成该事件的响应，也就是回调函数们。

从这个意义上看，我们所关注所谓的 订阅者 可以看作 一个事件发布后，发布者需要调用的一组函数。而所谓 发布，实际上就是调用这组函数以 完成事件（的回调）。因此我们接下来该用 listener 表示监听事件以待执行的回调函数， event 表示事件名， emit 表示这个事件发生后需要由发布者调用函数的过程。

至此，我们的 Publisher 从一个 Subscriber 的服务员 转型成为了职业 事件的管理者，不妨给它改个名——EventEmitter。

现在我们实现一个最基础的 EventEmitter 对象：

use std::collections::HashMap;

type EventCallback = Box<dyn Fn()>;

struct EventEmitter {
    /// 事件-监听器数组容器
    _events: HashMap<String, Vec<EventCallback>>,
}

impl EventEmitter {
    fn new() -> EventEmitter {
        EventEmitter {
            events: HashMap::new(),
        }
    }

	/// 添加事件监听器，监听器是一个回调函数，表示用户订阅的具体服务
    fn add_listener(&mut self, event: &str, callback: EventCallback) {
        let callbacks = self._events.entry(event.to_string()).or_insert(Vec::new());
        callbacks.push(callback);
    }
    
	/// 移除事件监听器：相当于用户取消订阅
    fn remove_listener(&mut self, event: &str, callback: &EventCallback) {
        if let Some(callbacks) = self._events.get_mut(event) {
            callbacks.retain(|cb| cb != callback);
        }
    }

	/// 触发事件：相当于发布消息或服务，也就是事件发生时，将订阅者订阅的服务一一为订阅者执行
    fn emit(&self, event: &str) {
        if let Some(callbacks) = self._events.get(event) {
            for callback in callbacks {
                callback();
            }
        }
    }
}

不过很多时候我们还不满足于此，比如能够限制监听器的数量。从现实生活中打个比方，就像我们只服务一定数量的客户，一旦订满，由于资源有限，不再接收其它订阅。更贴切实际地说，就像节假日你去旅游景区地酒店订房间，对于酒店来说，一旦所有房间都预定满了，就不再接收新的订阅了——除非，有已经订阅地客人退订了它们先前已经预定地房间。

实现这样一个功能，只需要一个变量 _max_listeners 作为最大监听器数量的控制变量。在外部相应的我们需要允许用户修改和读取该变量的值，因此还要提供 set_max_listeners 和 get_max_listeners 两个方法。

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

type EventCallback = Arc<dyn Fn() + Send + Sync>;

pub struct EventEmitter {
    _events: Mutex<HashMap<String, Vec<EventCallback>>>,
    _max_listeners: usize,
}

impl EventEmitter {
    pub fn new() -> Self {
        EventEmitter {
            _events: Mutex::new(HashMap::new()),
            _max_listeners: usize::MAX,
        }
    }

    /// 设置最大监听器数量
    /// Set the maximum number of listeners
    pub fn set_max_listeners(&mut self, max_listeners: usize) {
        self._max_listeners = max_listeners;
    }

    /// 获取最大监听器数量
    pub fn get_max_listeners(&self) -> usize {
        self._max_listeners
    }
    
    /// 添加事件监听器
    pub fn add_listener(&self, event: &str, callback: EventCallback) {
        let mut events = self._events.lock().unwrap();
        let callbacks = events.entry(event.to_string()).or_insert(Vec::new());
        callbacks.push(callback);
    }


	/// 移除事件监听器
    pub fn remove_listener(&self, event: &str, callback: &EventCallback) {
        let mut events = self._events.lock().unwrap();
        if let Some(callbacks) = events.get_mut(event) {
            callbacks.retain(|cb| !Arc::ptr_eq(cb, callback));
        }
    }
    /// 触发事件
    pub fn emit(&self, event: &str) {
        let events = self._events.lock().unwrap();
        if let Some(callbacks) = events.get(event) {
            for callback in callbacks {
                let callback_clone = callback.clone();
                // Spawn a new thread to run each callback asynchronously
                std::thread::spawn(move || {
                    (*callback_clone)();
                });
            }
        }
    }
}