EventBus源码阅读

示例代码

    // 注册一个处理String类型的监听
    @Subscribe(threadMode = ThreadMode.MAIN)
    public void onHandleMessage(String a) {
        mStater.setText(a);
    }
    // 注册一个处理Msg类型的监听（Msg是我随便写的一个类，只有 m ：String 的属性）
    @Subscribe(threadMode = ThreadMode.MAIN)
    public void onHandleMsg(Msg a) {
        mStater.setText(a.m);
    }

存储注册

• 存储消息处理器：subscriptionsByEventType

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

image.png

• 存储注册的类型 typesBySubscriber
  为了在注销监听时，将所有类型的的监听都注销掉。

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

image.png

消息分发

    public void post(Object event) {
        // 从ThreadLocal中获取当前Thread的消息分发队列
        PostingThreadState postingState = currentPostingThreadState.get();
        List<Object> eventQueue = postingState.eventQueue;
        eventQueue.add(event);
        // 当前线程的消息分发队列处于空闲时，则开始分发
        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 postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
        switch (subscription.subscriberMethod.threadMode) {
            case POSTING: // 不切线程，哪个线程调用的post，就在哪个线程处理消息
                invokeSubscriber(subscription, event);
                break;
            case MAIN: 
                if (isMainThread) { 
                    // 如果主线程调用的post，则直接调用方法出出力消息
                    invokeSubscriber(subscription, event);
                } else {
                    // 调用主线程Poster
                    // 实际是一个主线程的Handler内部维护一个PaddingPostQueue，内部的消息可以依次分发
                    mainThreadPoster.enqueue(subscription, event);
                }
                break;
            case MAIN_ORDERED:
                // 默认都走主线程Poster，因为内部的队列可以保证顺序
                if (mainThreadPoster != null) {
                    mainThreadPoster.enqueue(subscription, event);
                } else {
                    invokeSubscriber(subscription, event);
                }
                break;
            case BACKGROUND:
                if (isMainThread) {
                    // 如果是主线程，则入背后线程Poster
                    // 背后线程是一个Runnable，内部包含一个PendingPostQueue
                    // 当有消息进来时，判断Runnable状态，如果非激活状态，则调用EventBus的线程池执行该Runnable，依次处理队列内的消息，所以每次提交线程池时，消息处理是有序的
                    backgroundPoster.enqueue(subscription, event);
                } else {
                    // 如果不是主线程，哪里调用Post，就在哪里处理消息
                    invokeSubscriber(subscription, event);
                }
                break;
            case ASYNC:
                    // 背后线程是一个Runnable，内部包含一个PendingPostQueue
                    // 当有消息进来时，入队，然后调用EventBus线程池执行该Runnable，来一次消息，执行一次
                asyncPoster.enqueue(subscription, event);
                break;
            default:
                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
        }
    }

贴一个AsyncPoster的代码吧

class AsyncPoster implements Runnable, Poster {
    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);
    }
}