EventBus 使用/架构/源码分析

eventbus是针对android优化的发布-订阅事件总线，简化了android组件间的通信。eventbus以其简单易懂、优雅、开销小等优点而备受欢迎。

github 地址：https://github.com/greenrobot/eventbus

1. 使用

1.1 gradle中引入

        api 'org.greenrobot:eventbus:3.0.0'

1.2 定义事件

定义一个类作为事件，可以在类中定义不同的参数，发布者赋值，订阅者取值。

public class testevent {

    private string mname;


    public testevent(string name) {
        mname = name;
    }

    public string geteventname() {
        return mname;
    }
}

1.3 注册事件

首先需要将当前对象（activity/fragment等）与eventbus绑定（一般在oncreate函数中进行注册）

eventbus.getdefault().register(this);

接收事件的函数：

@subscribe (threadmode = threadmode.main, sticky = true)
public void ontestkeyevent(testevent event) {
        log.d(tag, "ontestkeyevent | eventname=" + event.geteventname());
        toast.maketext(this, "test event, name=" + event.geteventname(), toast.length_short).show();
    }

这里通过注解的方式，定义事件的类型，和回调的线程等信息。

查看eventbus jar包中subscribe定义：

@documented
@retention(retentionpolicy.runtime)
@target({elementtype.method})
public @interface subscribe {
    

threadmode

 threadmode() default threadmode.posting;

    /**
     * if true, delivers the most recent sticky event (posted with
     * {@link eventbus#poststicky(object)}) to this subscriber (if event available).
     */
    boolean 

sticky

() default false;

    /** subscriber priority to influence the order of event delivery.
     * within the same delivery thread ({@link threadmode}), higher priority subscribers will receive events before
     * others with a lower priority. the default priority is 0. note: the priority does *not* affect the order of
     * delivery among subscribers with different {@link threadmode}s! */
    int 

priority

() default 0;
}

 

查看eventbus jar包中threadmode定义：

a) posting : 回调在发布者线程

b) main : 回调在主线程

c) background : 回调在子线程（如果发布在子线程者回调直接运行在该线程）

d) async : 异步回调（不回回调在发布线程也不会回调在主线程）

 

1.4 发送事件

发布者不需要进行注册，只需要将事件post出去。

a)  普通事件：eventbus.getdefault().post(new testevent("normalevent"));

b） 粘性事件：eventbus.getdefault().poststicky(new testevent("stickevent"));

 

普通事件和粘性事件区别：

     如果发布的是普通事件，当前如果没有subscriber，则后续注册的subscriber也不会收到该事件。

     如果发布的是粘性事件，当前如果没有subscriber，内部会暂存该事件，当注册subscriber时，该subscriber会立刻收到该事件。

 

2. 结构

采用了典型的订阅发布设计模式。

3. 源码分析

// 这里只分析其原理和结构不会细细推敲每一行代码

订阅者信息封装（subscription）：

定义了两个成员变量，

final object subscriber;  // 订阅一个事件的对象
final subscribermethod subscribermethod; // 订阅的具体信息（方法名/threadmode/isstrick/priority）

eventbus主要成员变量：

private final map<class<?>, copyonwritearraylist<subscription>> subscriptionsbyeventtype;
private final map<object, list<class<?>>> typesbysubscriber;
private final map<class<?>, object> stickyevents;

subscriptionsbyeventtype:以event(即事件类)为key，以订阅列表(subscription)为value，事件发送之后，在这里寻找订阅者,而subscription又是一个copyonwritearraylist，这是一个线程安全的容器。你们看会好奇，subscription到底是什么，其实看下去就会了解的，现在先提前说下：subscription是一个封装类，封装了订阅者、订阅方法这两个类。
typesbysubscriber：以订阅者类为key，以event事件类为value，在进行register或unregister操作的时候，会操作这个map。
stickyevents:保存的是粘性事件

 

3.1 注册subscriber

注册过程，也就是调用regester函数的执行过程（主要是通过反射将注册者信息添加到上述讲的两个map中：typesbysubscriber、subscriptionsbyeventtype）

a) subscribermethodfinder 是专门用来查找目标对象中所有订阅函数（带缓存，避免同一个类多次反射查找）。反射可以获取函数的注解内容及每个函数的返回值/修饰符，具体查看findusingreflectioninsingleclass函数。

private void findusingreflectioninsingleclass(findstate findstate) {
        method[] methods;
        try {
            // this is faster than getmethods, especially when subscribers are fat classes like activities
            methods = findstate.clazz.getdeclaredmethods();
        } catch (throwable th) {
            // workaround for java.lang.noclassdeffounderror, see https://github.com/greenrobot/eventbus/issues/149
            methods = findstate.clazz.getmethods();
            findstate.skipsuperclasses = true;
        }
        for (method method : methods) {
            int modifiers = method.getmodifiers();
            if ((modifiers & modifier.public) != 0 && (modifiers & modifiers_ignore) == 0) {
                class<?>[] parametertypes = method.getparametertypes();
                if (parametertypes.length == 1) {
                    subscribe subscribeannotation = method.getannotation(subscribe.class);
                    if (subscribeannotation != null) {
                        class<?> eventtype = parametertypes[0];
                        if (findstate.checkadd(method, eventtype)) {
                            threadmode threadmode = subscribeannotation.threadmode();
                            findstate.subscribermethods.add(new subscribermethod(method, eventtype, threadmode,
                                    subscribeannotation.priority(), subscribeannotation.sticky()));
                        }
                    }
                } else if (strictmethodverification && method.isannotationpresent(subscribe.class)) {
                    string methodname = method.getdeclaringclass().getname() + "." + method.getname();
                    throw new eventbusexception("@subscribe method " + methodname +
                            "must have exactly 1 parameter but has " + parametertypes.length);
                }
            } else if (strictmethodverification && method.isannotationpresent(subscribe.class)) {
                string methodname = method.getdeclaringclass().getname() + "." + method.getname();
                throw new eventbusexception(methodname +
                        " is a illegal @subscribe method: must be public, non-static, and non-abstract");
            }
        }
    }

b) 将订阅函数添加到两个缓存map中

c) 如果订阅函数接收的是粘性事件，则将缓存中的粘性事件回调给该订阅函数。

上述b) c) 两个步骤的具体代码如下：

private void subscribe(object subscriber, subscribermethod subscribermethod) {
        class<?> eventtype = subscribermethod.eventtype;
        subscription newsubscription = new subscription(subscriber, subscribermethod);
        copyonwritearraylist<subscription> subscriptions = subscriptionsbyeventtype.get(eventtype);
        if (subscriptions == null) {
            subscriptions = new copyonwritearraylist<>();
            subscriptionsbyeventtype.put(eventtype, subscriptions);
        } else {
            if (subscriptions.contains(newsubscription)) {
                throw new eventbusexception("subscriber " + subscriber.getclass() + " already registered to event "
                        + eventtype);
            }
        }

        int size = subscriptions.size();
        for (int i = 0; i <= size; i++) {
            if (i == size || subscribermethod.priority > subscriptions.get(i).subscribermethod.priority) {
                subscriptions.add(i, newsubscription);
                break;
            }
        }

        list<class<?>> subscribedevents = typesbysubscriber.get(subscriber);
        if (subscribedevents == null) {
            subscribedevents = new arraylist<>();
            typesbysubscriber.put(subscriber, subscribedevents);
        }
        subscribedevents.add(eventtype);

        if (subscribermethod.sticky) {
            if (eventinheritance) {
                // existing sticky events of all subclasses of eventtype have to be considered.
                // note: iterating over all events may be inefficient with lots of sticky events,
                // thus data structure should be changed to allow a more efficient lookup
                // (e.g. an additional map storing sub classes of super classes: class -> list<class>).
                set<map.entry<class<?>, object>> entries = stickyevents.entryset();
                for (map.entry<class<?>, object> entry : entries) {
                    class<?> candidateeventtype = entry.getkey();
                    if (eventtype.isassignablefrom(candidateeventtype)) {
                        object stickyevent = entry.getvalue();
                        checkpoststickyeventtosubscription(newsubscription, stickyevent);
                    }
                }
            } else {
                object stickyevent = stickyevents.get(eventtype);
                checkpoststickyeventtosubscription(newsubscription, stickyevent);
            }
        }
    }

3.2 分发消息到每个subscriber

分发过程是从subscriptionsbyeventtype中取subscriber并在指定的线程中回调接收函数的过程。

如何实现在不同线程中执行回调函数？

a）从订阅信息中获取订阅函数回调线程。

b) 在指定线程中回调订阅函数。

分发消息过程

private void posttosubscription(subscription subscription, object event, boolean ismainthread) {
        switch (subscription.subscribermethod.threadmode) {
            case posting:
                invokesubscriber(subscription, event);
                break;
            case main:
                if (ismainthread) {
                    invokesubscriber(subscription, event);
                } else {
                    mainthreadposter.enqueue(subscription, event);
                }
                break;
            case background:
                if (ismainthread) {
                    backgroundposter.enqueue(subscription, event);
                } else {
                    invokesubscriber(subscription, event);
                }
                break;
            case async:
                asyncposter.enqueue(subscription, event);
                break;
            default:
                throw new illegalstateexception("unknown thread mode: " + subscription.subscribermethod.threadmode);
        }
    }

不同的的消息分发器在eventbus构造的时候初始化，下面看一下asyncposter的源码如下：

class asyncposter implements runnable {

    private final pendingpostqueue queue;
    private final eventbus eventbus;

    asyncposter(eventbus eventbus) {
        this.eventbus = eventbus;
        queue = new pendingpostqueue();
    }

    public void enqueue(subscription subscription, object event) {
        pendingpost pendingpost = pendingpost.obtainpendingpost(subscription, event);
        queue.enqueue(pendingpost);
        eventbus.getexecutorservice().execute(this);
    }

    @override
    public void run() {
        pendingpost pendingpost = queue.poll();
        if(pendingpost == null) {
            throw new illegalstateexception("no pending post available");
        }
        eventbus.invokesubscriber(pendingpost);
    }

}

asyncposter分发器继承自runable，核心是通过自定义的阻塞队列维护消息，然后在eventbus定义的线程池中执行runable接口中的代码。

eventbus中还定义了backgroundposter/handlerposter这里不赘述。

3.3 物理类图

 

其它细节：

上述分析只是讲解了eventbus大概原理，并没有细细分析。如，代码中很多考虑了并发，事件优先级等