如何自己实现事件的订阅和发布呢？

1.原理

核心思想是基于发布/订阅模式，用一个共享的数据结构来管理事件和事件监听器。主要功能包括事件订阅、取消订阅、发布事件等功能。

实现思路

1. 定义事件和监听器接口：首先定义一个 Event 类和一个 EventListener 接口，所有事件和监听器都继承自它们。
2. 管理订阅关系：使用 Map<Class<? extends Event>, List<EventListener<?>>> 来存储事件类型与监听器的关系。
3. 发布事件：通过指定事件类型查找所有订阅的监听器，并调用监听器的处理方法。

2.代码工程

Step 1: 定义基础的事件和监听器接口

package com.et;

// base event
public abstract class Event {
    // some basic filed
}

package com.et;

// event listen interface
public interface EventListener<T extends Event> {
    void onEvent(T event);
}

Step 2: 实现 EventBus 类

package com.et;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class EventBus {
    // store event
    private final Map<Class<? extends Event>, List<EventListener<? extends Event>>> listeners = new HashMap<>();

    // Subscribe
    public <T extends Event> void register(Class<T> eventType, EventListener<T> listener) {
        listeners.computeIfAbsent(eventType, k -> new ArrayList<>()).add(listener);
    }

    // cancel Subscribe
    public <T extends Event> void unregister(Class<T> eventType, EventListener<T> listener) {
        List<EventListener<? extends Event>> eventListeners = listeners.get(eventType);
        if (eventListeners != null) {
            eventListeners.remove(listener);
        }
    }

    // publish
    @SuppressWarnings("unchecked")
    public <T extends Event> void post(T event) {
        List<EventListener<? extends Event>> eventListeners = listeners.get(event.getClass());
        if (eventListeners != null) {
            for (EventListener<? extends Event> listener : eventListeners) {
                ((EventListener<T>) listener).onEvent(event); // event forward
            }
        }
    }
}

Step 3: 定义自定义事件和监听器

package com.et;

// UserLoginEvent
public class UserLoginEvent extends Event {
    private final String username;

    public UserLoginEvent(String username) {
        this.username = username;
    }

    public String getUsername() {
        return username;
    }
}

package com.et;

public class UserLoginListener implements EventListener<UserLoginEvent> {
    @Override
    public void onEvent(UserLoginEvent event) {
        System.out.println("User logged in: " + event.getUsername());
    }
}

代码解释

1. register 方法：将监听器与事件类型绑定。
2. post 方法：发布事件，找到所有订阅该事件的监听器并通知它们。
3. unregister 方法：从事件类型的监听器列表中移除指定监听器。

优化建议

1. 异步处理：可以使用线程池异步发布事件，提高性能。
2. 过滤器机制：支持对事件的过滤，以便监听器只接收满足条件的事件。
3. 事件优先级：可以为监听器定义优先级，根据优先级顺序处理。

这种简单的 EventBus 实现适合单进程内的轻量事件传递。如果需要更复杂的功能（如分布式支持、持久化消息），则可以使用成熟的消息代理系统，如 RabbitMQ 或 Kafka。

以上只是一些关键代码，所有代码请参见下面代码仓库

代码仓库

• GitHub - Harries/Java-demo(customeMq)

3.测试

package com.et;

public class EventBusTest {
    public static void main(String[] args) {
        EventBus eventBus = new EventBus();
        
        // register
        UserLoginListener userLoginListener = new UserLoginListener();
        eventBus.register(UserLoginEvent.class, userLoginListener);
        
        // publish
        UserLoginEvent loginEvent = new UserLoginEvent("Alice");
        eventBus.post(loginEvent);
        
        // unregister
        eventBus.unregister(UserLoginEvent.class, userLoginListener);
        eventBus.post(new UserLoginEvent("Bob")); // don't take effort,because the event is unregister
    }
}

运行代码，返回结果

User logged in: Alice