Java?EventBus手把手帶你實(shí)現(xiàn)

更新時(shí)間：2023年01月09日 16:08:50 作者：鯤鵬飛九萬里

EventBus是Guava的事件處理機(jī)制，是設(shè)計(jì)模式中觀察者模式（生產(chǎn)/消費(fèi)者編程模型）的優(yōu)雅實(shí)現(xiàn)。本文就來和大家聊聊EventBus的使用，需要的可以參考一下

一、說明

在Guava中，EventBus簡(jiǎn)化了觀察者模式的實(shí)現(xiàn)。理解EventBus的原理來，自動(dòng)動(dòng)手實(shí)現(xiàn)一個(gè)簡(jiǎn)單的EventBus。

二、Guava的EventBus

EventBus叫做“時(shí)間總線”，它提供了實(shí)現(xiàn)觀察者模式的骨架代碼?？梢曰诖丝蚣埽浅Ｈ菀椎卦谧约旱臉I(yè)務(wù)場(chǎng)景中實(shí)現(xiàn)觀察者模式。它不僅支持異步非阻塞模式，同時(shí)支持同步阻塞模式。

基于EventBus，不需要定義Observer接口（觀察者接口），任意類型的對(duì)象都可以注冊(cè)到EventBus中。通過@Subscribe注解來表明類中哪個(gè)函數(shù)可以接收觀察者發(fā)送的消息。

Guava EventBus中的幾個(gè)主要的類和函數(shù)：

EventBus、SyncEventBus

EventBus類中封裝了對(duì)外暴露的所有可調(diào)用接口。其中EventBus實(shí)現(xiàn)了同步阻塞的觀察者模式，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() 函數(shù)

EventBus通過register()函數(shù)來注冊(cè)觀察者。它可以接收任意類型（Object）的觀察者。具體的函數(shù)定義如下：

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

unregister() 函數(shù)

相對(duì)于register()，unregister()函數(shù)是從EventBus中刪除某個(gè)觀察者。

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

post函數(shù)

EventBus提供post()函數(shù) ，用來給觀察者發(fā)消息。

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

post發(fā)送消息的時(shí)候，并不是把消息發(fā)送給所有的觀察者，而是發(fā)送給可匹配的觀察者。所謂可匹配指的是，能接收的消息類型是發(fā)送消息（post函數(shù)中定義的父類）。

比如，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注解類標(biāo)明，某個(gè)函數(shù)能接收哪種類型的消息。（類型不能是基本類型）

三、EventBus的原理

四、動(dòng)手實(shí)現(xiàn)一個(gè)EventBus

@Beat 標(biāo)注一個(gè)公共的API（公共的類、方法或字段）在未來的發(fā)行版本中會(huì)發(fā)生不兼容的變化。帶有此注釋的 API 不受其包含庫所做的任何兼容性保證。請(qǐng)注意，此注釋的存在并不意味著所討論 API 的質(zhì)量或性能，只是它不是“API 凍結(jié)”的事實(shí)。

應(yīng)用程序依賴 beta API 通常是安全的，但需要在升級(jí)期間進(jìn)行一些額外的工作。然而，不建議在類庫（包含在用戶的CLASSPATH中，不受開發(fā)人員的控制）上這么做。

4.1 定義Subscribe注解

定義Subscribe注解，用于標(biāo)明哪個(gè)函數(shù)可以接收消息。

/**
 * 定義一個(gè)注解，表明觀察者中的哪個(gè)函數(shù)可以接收消息
 */
@Retention(RetentionPolicy.RUNTIME)  // 注解的聲明周期
@Target(ElementType.METHOD)  // 注解作用的地方
@Beta  // 標(biāo)注API在未來發(fā)行的版本是可能有不兼容的變化
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方法的參數(shù)
     * @param event
     */
    public void execute(Object event) {
        try {
            method.invoke(target, event);
        } catch (IllegalAccessException | InvocationTargetException  e) {
            throw new RuntimeException(e);
        }
    }
}

4.3 ObserverRegister

Observer 注冊(cè)表。

/**
 * Observer 注冊(cè)表
 */
public class MyObserverRegister {
    // 注冊(cè)表, 消息類型: 觀察者方法
    private ConcurrentMap<Class<?>, CopyOnWriteArraySet<MyObserverAction>> registry = new ConcurrentHashMap<>();
    /**
     * 將觀察者注冊(cè)到 注冊(cè)表中
     * @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;
    }
    /**
     * 根據(jù)觀察者類型，查找方法列表
     * @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 有一個(gè)注解@MySubscribe ,它有%s個(gè)參數(shù)，實(shí)際要求有且只有一個(gè)參數(shù)",
                        method, parameterTypes.length);
                result.add(method);
            }
        }
        return result;
    }
}

4.4 EventBus

/**
 * 實(shí)現(xiàn) 同步阻塞的 EventBus
 */
public class MyEventBus {
    private Executor executor;
    private MyObserverRegister register = new MyObserverRegister();
    public MyEventBus() {
        // MoreExecutors.directExecutor() 是 Google Guava 提供的工具類，看似是多線程，實(shí)際上是單線程。
        // 之所以要這么實(shí)現(xiàn)，主要還是為了跟 AsyncEventBus 統(tǒng)一代碼邏輯，做到代碼復(fù)用
        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();
        // 注冊(cè)一個(gè)觀察者
        myEventBus.register(new CurrentConditionsDisplayListener());
        // 向觀察者發(fā)送消息
        myEventBus.post(23.0f);
    }