Java EventBus手把手带你实现

// 同步阻塞的方式
EventBus eventBus = new EventBus(); 
// 异步非阻塞的方式
final int DEFAULT_EVENTBUS_THREAD_POOL_SIZE = 20; // 异步非阻塞线程池大小
ExecutorService executorService = Executors.newFixedThreadPool(DEFAULT_EVENTBUS_THREAD_POOL_SIZE);
EventBus eventBus = new AsyncEventBus(executorService);

public void register(Object object) {
  //......
}

public void unregister(Object object) {
  //......
}

public void post(Object event) {
  //......
}

XMsg xMsg= new XMsg();
YMsg yMsg= new YMsg();
ZMsg zMsg= new ZMsg();
post(xMsg);// AObserver  接收消息
post(yMsg);// AObserver和BObserver接收到消息
post(zMsg);// CObserver接收到消息

/**
 * 定义一个注解，表明观察者中的哪个函数可以接收消息
 */
@Retention(RetentionPolicy.RUNTIME)  // 注解的声明周期
@Target(ElementType.METHOD)  // 注解作用的地方
@Beta  // 标注API在未来发行的版本是可能有不兼容的变化
public @interface MySubscribe {
}

/**
 * 用来表示 @MySubscribe 注解方法
 */
public class MyObserverAction {
    private Object target;
    private Method method;
    public MyObserverAction(Object target, Method method) {
        this.target = checkNotNull(target);
        this.method = method;
        this.method.setAccessible(true);
    }
    /**
     * event是method方法的参数
     * @param event
     */
    public void execute(Object event) {
        try {
            method.invoke(target, event);
        } catch (IllegalAccessException | InvocationTargetException  e) {
            throw new RuntimeException(e);
        }
    }
}

/**
 * Observer 注册表
 */
public class MyObserverRegister {
    // 注册表, 消息类型: 观察者方法
    private ConcurrentMap<Class<?>, CopyOnWriteArraySet<MyObserverAction>> registry = new ConcurrentHashMap<>();
    /**
     * 将观察者注册到 注册表中
     * @param observer 观察者
     */
    public void register(Object observer) {
        Map<Class<?>, Collection<MyObserverAction>> observerActions = findAllObserverActions(observer);
        for (Map.Entry<Class<?>, Collection<MyObserverAction>> entry : observerActions.entrySet()) {
            Class<?> eventType = entry.getKey();
            Collection<MyObserverAction> evenActions = entry.getValue();
            CopyOnWriteArraySet<MyObserverAction> registryEvenActions =
                    registry.getOrDefault(eventType, new CopyOnWriteArraySet<>());
            registryEvenActions.addAll(evenActions);
            registry.put(eventType, registryEvenActions);
        }
    }
    /**
     * 获取匹配的观察者事件
     * @param event
     * @return
     */
    public List<MyObserverAction> getMatchedMyObserverActions(Object event) {
        List<MyObserverAction> result = new ArrayList<>();
        Class<?> postedEventClass = event.getClass();
        for (Map.Entry<Class<?>, CopyOnWriteArraySet<MyObserverAction>> entry : registry.entrySet()) {
            Class<?> eventClass = entry.getKey();
            // 匹配相同类型或父类型
            if (postedEventClass.isAssignableFrom(eventClass)) {
                result.addAll(entry.getValue());
            }
        }
        return result;
    }
    // 消息类型（观察者类型类型及其父类型） 观察者方法
    public Map<Class<?>, Collection<MyObserverAction>> findAllObserverActions(Object observer) {
        Map<Class<?>, Collection<MyObserverAction>> result = new HashMap<>();
        // 观察者类型
        Class<?> observerClass = observer.getClass();
        for (Method method : getAnnotatedMethods(observerClass)) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            Class<?> eventType = parameterTypes[0];
            result.putIfAbsent(eventType, new ArrayList<>());
            result.get(eventType).add(new MyObserverAction(observer, method));
        }
        return result;
    }
    /**
     * 根据观察者类型，查找方法列表
     * @param clazz
     * @return
     */
    public List<Method> getAnnotatedMethods(Class<?> clazz) {
        List<Method> result = new ArrayList<>();
        for (Method method : clazz.getDeclaredMethods()) {
            if (method.isAnnotationPresent(MySubscribe.class)) {
                Class<?>[] parameterTypes = method.getParameterTypes();
                checkArgument(parameterTypes.length==1,
                        "方法%s 有一个注解@MySubscribe ,它有%s个参数，实际要求有且只有一个参数",
                        method, parameterTypes.length);
                result.add(method);
            }
        }
        return result;
    }
}

/**
 * 实现 同步阻塞的 EventBus
 */
public class MyEventBus {
    private Executor executor;
    private MyObserverRegister register = new MyObserverRegister();
    public MyEventBus() {
        // MoreExecutors.directExecutor() 是 Google Guava 提供的工具类，看似是多线程，实际上是单线程。
        // 之所以要这么实现，主要还是为了跟 AsyncEventBus 统一代码逻辑，做到代码复用
        this(MoreExecutors.directExecutor());
    }
    // 注意这里的修饰符
    protected MyEventBus(Executor executor) {
        this.executor = executor;
    }
    public void register(Object observer) {
        register.register(observer);
    }
    public void post(Object event) {
        List<MyObserverAction> observerActions = register.getMatchedMyObserverActions(event);
        for (MyObserverAction observerAction : observerActions) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    observerAction.execute(event);
                }
            });
        }
    }
}

/**
 * 异步非阻塞的EventBus
 */
public class MySyncEventBus extends MyEventBus {
    public MySyncEventBus(Executor executor) {
        super(executor);
    }
}

    public static void main(String[] args) {
        // 自定义的EventBus
        MyEventBus myEventBus = new MyEventBus();
        // 注册一个观察者
        myEventBus.register(new CurrentConditionsDisplayListener());
        // 向观察者发送消息
        myEventBus.post(23.0f);
    }