其他分享
首页 > 其他分享> > EventBus

EventBus

作者：互联网

EventBus配置

android {
    defaultConfig {
        javaCompileOptions {
            annotationProcessorOptions {
                // 生成的Index类的名称
                arguments = [ eventBusIndex : 'com.example.myapp.MyEventBusIndex' ]
            }
        }
    }
}

dependencies {
    def eventbus_version = '3.2.0'
    implementation "org.greenrobot:eventbus:$eventbus_version"
    annotationProcessor "org.greenrobot:eventbus-annotation-processor:$eventbus_version"
}

EventBus使用流程

EventBus中的数据结构

Map变量：subscriptionsByEventType

//eventType : Subscription
private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;

Map变量：typesBySubscriber

private final Map<Object, List<Class<?>>> typesBySubscriber;

Subscription.java

final class Subscription {
    final Object subscriber; // Object对象，反射invoke方法时传入
    final SubscriberMethod subscriberMethod; // 一个使用Subscribe注解标记的方法
}

SubscriberMethod.java

final Method method; //经过注册解析得到的函数方法，用于处理post的事件。
final ThreadMode threadMode; // 线程模式
final Class<?> eventType; // event 事件类型。注解方法的入参类型，post函数的入参类型
final int priority; // 优先级，优先级越高，越先收到消息通知。
final boolean sticky; // 是否支持粘性方法。

PendingPost.java

PendingPostQueue.java

注解相关

@Subscribe注解

ThreadMode

    org.greenrobot.eventbus.EventBusBuilder
    private final static ExecutorService DEFAULT_EXECUTOR_SERVICE = Executors.newCachedThreadPool();


/**
 * Each subscriber method has a thread mode, which determines in which thread the method is to be called by EventBus.
 * EventBus takes care of threading independently from the posting thread.
 *
 * @see EventBus#register(Object)
 * @author Markus
 */
public enum ThreadMode {
    /**
     * Subscriber will be called directly in the same thread, which is posting the event. This is the default. Event delivery
     * implies the least overhead because it avoids thread switching completely. Thus this is the recommended mode for
     * simple tasks that are known to complete in a very short time without requiring the main thread. Event handlers
     * using this mode must return quickly to avoid blocking the posting thread, which may be the main thread.
     */
    POSTING,

    /**
     * On Android, subscriber will be called in Android's main thread (UI thread). If the posting thread is
     * the main thread, subscriber methods will be called directly, blocking the posting thread. Otherwise the event
     * is queued for delivery (non-blocking). Subscribers using this mode must return quickly to avoid blocking the main thread.
     * If not on Android, behaves the same as {@link #POSTING}.
     */
    MAIN,

    /**
     * On Android, subscriber will be called in Android's main thread (UI thread). Different from {@link #MAIN},
     * the event will always be queued for delivery. This ensures that the post call is non-blocking.
     */
    MAIN_ORDERED,

    /**
     * On Android, subscriber will be called in a background thread. If posting thread is not the main thread, subscriber methods
     * will be called directly in the posting thread. If the posting thread is the main thread, EventBus uses a single
     * background thread, that will deliver all its events sequentially. Subscribers using this mode should try to
     * return quickly to avoid blocking the background thread. If not on Android, always uses a background thread.
     */
    BACKGROUND,

    /**
     * Subscriber will be called in a separate thread. This is always independent from the posting thread and the
     * main thread. Posting events never wait for subscriber methods using this mode. Subscriber methods should
     * use this mode if their execution might take some time, e.g. for network access. Avoid triggering a large number
     * of long running asynchronous subscriber methods at the same time to limit the number of concurrent threads. EventBus
     * uses a thread pool to efficiently reuse threads from completed asynchronous subscriber notifications.
     */
    ASYNC
}

初始化注册流程

register 方法

/**
 * Registers the given subscriber to receive events. Subscribers must call {@link #unregister(Object)} once they
 * are no longer interested in receiving events.
 * <p/>
 * Subscribers have event handling methods that must be annotated by {@link Subscribe}.
 * The {@link Subscribe} annotation also allows configuration like {@link
 * ThreadMode} and priority.
 */
public void register(Object subscriber) {
    Class<?> subscriberClass = subscriber.getClass();
    // 解析类的所有方法，找到有注解的方法
    List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
    synchronized (this) {
        for (SubscriberMethod subscriberMethod : subscriberMethods) {
            subscribe(subscriber, subscriberMethod); //将解析的方法，添加到map数据结构中去
        }
    }
}

findSubscriberMethods，解析@Subscribe解方法

List<SubscriberMethod> findSubscriberMethods(Class<?> subscriberClass) {
    List<SubscriberMethod> subscriberMethods = METHOD_CACHE.get(subscriberClass); //查缓存
    if (subscriberMethods != null) {
        return subscriberMethods;
    }

    if (ignoreGeneratedIndex) { // 是否忽略通过预编译生成的Index查找Method额方式，默认值false
        subscriberMethods = findUsingReflection(subscriberClass); //直接通过反射查找类
    } else {
        subscriberMethods = findUsingInfo(subscriberClass); // 查找注解方法
    }
    if (subscriberMethods.isEmpty()) {
        throw new EventBusException("Subscriber " + subscriberClass
                + " and its super classes have no public methods with the @Subscribe annotation");
    } else {
        METHOD_CACHE.put(subscriberClass, subscriberMethods);
        return subscriberMethods;
    }
}

private List<SubscriberMethod> findUsingInfo(Class<?> subscriberClass) {
    FindState findState = prepareFindState(); // 初始化了一个对象池，对象池大小4
    findState.initForSubscriber(subscriberClass);
    while (findState.clazz != null) { // while循环遍历父类
        // 通过注解解析器在编译期间生成注解，避免通过反射查找，来加快查找速度。
        // 可以参考 SimpleSubscriberInfo
        findState.subscriberInfo = getSubscriberInfo(findState);
        if (findState.subscriberInfo != null) {
            SubscriberMethod[] array = findState.subscriberInfo.getSubscriberMethods();
            for (SubscriberMethod subscriberMethod : array) {
                // 校验正确性。这个函数里面限制了：如果父类和子类同时注解了一个同一个方法，那么只有子类生效
                if (findState.checkAdd(subscriberMethod.method, subscriberMethod.eventType)) {
                    findState.subscriberMethods.add(subscriberMethod);
                }
            }
        } else {
            findUsingReflectionInSingleClass(findState); // 通过反射来查找Method
        }
        findState.moveToSuperclass(); //向上查找父类。此处通过类的包名过滤，忽略了Android的系统类，java的系统类，提升查找速度。
    }
    return getMethodsAndRelease(findState);
}

private static final int BRIDGE = 0x40; //桥接方法
private static final int SYNTHETIC = 0x1000; // 编译合成方法，编译器生成的方法
private static final int MODIFIERS_IGNORE = Modifier.ABSTRACT | Modifier.STATIC | BRIDGE | SYNTHETIC;

// 使用反射解析一个类
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) {
        // 省略 catch代码....
        findState.skipSuperClasses = true;
    }
    for (Method method : methods) {
        int modifiers = method.getModifiers();
        // 只解析public方法，并且忽略抽象、静态方法，忽略编译器生成的方法
        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();
                        // 构造 SubscriberMethod，添加到list列表
                        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);
            }
        // strictMethodVerification 是否启用严格校验，默认值false。
        // 在严格校验的情况下，如果错误的使用Subscribe注解方法，就会抛出异常
        } 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");
        }
    }
}

post 发送消息

ThreadLocal对象： currentPostingThreadState

    private final ThreadLocal<PostingThreadState> currentPostingThreadState = 
      new ThreadLocal<PostingThreadState>() {
        @Override
        protected PostingThreadState initialValue() {
            return new PostingThreadState();
        }
    };

post函数

/** Posts the given event to the event bus. */
public void post(Object event) {
    PostingThreadState postingState = currentPostingThreadState.get(); // ThreadLocal对象初始化
    List<Object> eventQueue = postingState.eventQueue;
    eventQueue.add(event); //post的事件添加到队列中

    if (!postingState.isPosting) { // 如果当前正在对外发送消息。
        postingState.isMainThread = isMainThread();
        postingState.isPosting = true;
        if (postingState.canceled) {
            throw new EventBusException("Internal error. Abort state was not reset");
        }
        try {
            while (!eventQueue.isEmpty()) {
                postSingleEvent(eventQueue.remove(0), postingState);
            }
        } finally {
            postingState.isPosting = false;
            postingState.isMainThread = false;
        }
    }
}

private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
    Class<?> eventClass = event.getClass();
    boolean subscriptionFound = false;
    if (eventInheritance) { // 是否考虑父类， 默认为true
        List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
        int countTypes = eventTypes.size();
        for (int h = 0; h < countTypes; h++) {
            Class<?> clazz = eventTypes.get(h);
            subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
        }
    } else {
        // 发送事件
        subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
    }
    if (!subscriptionFound) {
      // 省略.... 没找到事件的异常处理
    }
}

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
    CopyOnWriteArrayList<Subscription> subscriptions;
    synchronized (this) {
        subscriptions = subscriptionsByEventType.get(eventClass); // 从map中获取所有监听event type类型的订阅者
    }
    if (subscriptions != null && !subscriptions.isEmpty()) {
        for (Subscription subscription : subscriptions) { // for循环进行遍历操作
            postingState.event = event;
            postingState.subscription = subscription; // 订阅者赋值
            boolean aborted;
            try {
                postToSubscription(subscription, event, postingState.isMainThread);
                aborted = postingState.canceled;
            } finally {
                postingState.event = null;
                postingState.subscription = null;
                postingState.canceled = false;
            }
            if (aborted) {
                break;
            }
        }
        return true;
    }
    return false;
}

// 根据不同的threadMode，通过不同的线程进行处理
// 可以参考对ThreadMode的注解
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 MAIN_ORDERED:
            if (mainThreadPoster != null) { // Android场景下，这个变量一直非空
                mainThreadPoster.enqueue(subscription, event); // 送到主线程的队列调用
            } else {
                // temporary: technically not correct as poster not decoupled from subscriber
                invokeSubscriber(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);
    }
}

/**
 * Invokes the subscriber if the subscriptions is still active. Skipping subscriptions prevents race conditions
 * between {@link #unregister(Object)} and event delivery. Otherwise the event might be delivered after the
 * subscriber unregistered. This is particularly important for main thread delivery and registrations bound to the
 * live cycle of an Activity or Fragment.
 */
void invokeSubscriber(PendingPost pendingPost) {
    Object event = pendingPost.event;
    Subscription subscription = pendingPost.subscription;
    PendingPost.releasePendingPost(pendingPost);
    if (subscription.active) {
        invokeSubscriber(subscription, event);
    }
}

void invokeSubscriber(Subscription subscription, Object event) {
    try {
        subscription.subscriberMethod.method.invoke(subscription.subscriber, event); // 通过反射调用函数接口
    } catch (InvocationTargetException e) {
        handleSubscriberException(subscription, event, e.getCause());
    } catch (IllegalAccessException e) {
        throw new IllegalStateException("Unexpected exception", e);
    }
}

几种poster

EventBusAnnotationProcessor 注解解析器

// Build the project at least once to generate the index class specified with eventBusIndex.
// Then, e.g. in your Application class, use EventBus.builder().addIndex(indexInstance) to pass an instance of the index class to EventBus.
EventBus eventBus = EventBus.builder().addIndex(new MyEventBusIndex()).build();

// Use EventBusBuilder.installDefaultEventBus() to set the EventBus with index as the instance returned by EventBus.getDefault().
EventBus.builder().addIndex(new MyEventBusIndex()).installDefaultEventBus();
// Now the default instance uses the given index. Use it like this:
EventBus eventBus = EventBus.getDefault();

参考资料

标签：thread,subscriber,Subscribe,注解,EventBus,event,subscription
来源： https://blog.csdn.net/followYouself/article/details/120187507