示例代码
// 注册一个处理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);
}
}
