当前位置：主页 > java教程 > Java EventBus

Java EventBus手把手带你实现

发布：2023-03-04 13:00:01 59

给网友朋友们带来一篇相关的编程文章，网友池忆梅根据主题投稿了本篇教程内容，涉及到Java、EventBus、Java、EventBus实现、Java EventBus相关内容，已被150网友关注，如果对知识点想更进一步了解可以在下方电子资料中获取。

Java EventBus

一、说明

在Guava中，EventBus简化了观察者模式的实现。理解EventBus的原理来，自动动手实现一个简单的EventBus。

二、Guava的EventBus

EventBus叫做“时间总线”，它提供了实现观察者模式的骨架代码。可以基于此框架，非常容易地在自己的业务场景中实现观察者模式。它不仅支持异步非阻塞模式，同时支持同步阻塞模式。

基于EventBus，不需要定义Observer接口（观察者接口），任意类型的对象都可以注册到EventBus中。通过@Subscribe注解来表明类中哪个函数可以接收观察者发送的消息。

Guava EventBus中的几个主要的类和函数：

EventBus、SyncEventBus

EventBus类中封装了对外暴露的所有可调用接口。其中EventBus实现了同步阻塞的观察者模式，SyncEventBus继承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);

register() 函数

EventBus通过register()函数来注册观察者。它可以接收任意类型（Object）的观察者。具体的函数定义如下：

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

unregister() 函数

相对于register()，unregister()函数是从EventBus中删除某个观察者。

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

post函数

EventBus提供post()函数，用来给观察者发消息。

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

post发送消息的时候，并不是把消息发送给所有的观察者，而是发送给可匹配的观察者。所谓可匹配指的是，能接收的消息类型是发送消息（post函数中定义的父类）。

比如，AObserver能接收的消息类型是XMsg，BObserver能接收的消息类型是YMsg，CObserver能接收的消息类型是ZMsg。其中，XMsg是YMsg的父类。

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

Observer（观察者）能接收到消息类型是通过@Subscribe注解定义的。

@Subscribe 注解

EventBus通过@Subscribe注解类标明，某个函数能接收哪种类型的消息。（类型不能是基本类型）

三、EventBus的原理

四、动手实现一个EventBus

@Beat 标注一个公共的API（公共的类、方法或字段）在未来的发行版本中会发生不兼容的变化。带有此注释的 API 不受其包含库所做的任何兼容性保证。请注意，此注释的存在并不意味着所讨论 API 的质量或性能，只是它不是“API 冻结”的事实。

应用程序依赖 beta API 通常是安全的，但需要在升级期间进行一些额外的工作。然而，不建议在类库（包含在用户的CLASSPATH中，不受开发人员的控制）上这么做。

4.1 定义Subscribe注解

定义Subscribe注解，用于标明哪个函数可以接收消息。

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

4.2 ObserverAction

用来表示@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);
        }
    }
}

4.3 ObserverRegister

Observer 注册表。

/**
 * 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;
    }
}

4.4 EventBus

/**
 * 实现 同步阻塞的 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);
                }
            });
        }
    }
}

4.5 SyncEventBus

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

五、使用自定义的EventBus

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