前言
在前面的文章中，曾经提及过EventBus的几个重要成员，其中就包括以下这些：
private final Poster mainThreadPoster;
private final BackgroundPoster backgroundPoster;
private final AsyncPoster asyncPoster;
对于使用过EventBus的读者，应该可以理解这几个对象的作用，它们就是用来切换线程的。今天，我们就来详细分析这些对象。
一、Poster
BackgroundPoster 和AsyncPoster都是Poster的实现类，因此，需要先来分析一下这个顶级接口。
/**
* Posts events.
*
* @author William Ferguson
*/
interface Poster {
/**
* Enqueue an event to be posted for a particular subscription.
*
* @param subscription Subscription which will receive the event.
* @param event Event that will be posted to subscribers.
*/
void enqueue(Subscription subscription, Object event);
}
Poster接口的代码非常简单，加上注释也才30行左右。正如注释所言，该接口的作用就是发布事件的。Poster接口里只有一个enqueue方法，用于将事件入队处理。它接收两个参数：订阅信息和事件。要知道enqueue的具体逻辑，必然要到其实现类中查找。
二、HandlerPoster
mainThreadPoster在EventBus中比较独特，它是通过MainThreadSupport的内部方法createPoster创建的。
mainThreadPoster = mainThreadSupport != null ? mainThreadSupport.createPoster(this) : null;
MainThreadSupport也是一个接口类，因此，需要到其实现类查找createPoster的实现。遍寻EventBus后，发现其唯一实现类是AndroidHandlerMainThreadSupport，自然，这就是我们的目标了。
class AndroidHandlerMainThreadSupport implements MainThreadSupport {
private final Looper looper;
public AndroidHandlerMainThreadSupport(Looper looper) {
this.looper = looper;
}
@Override
public boolean isMainThread() {
return looper == Looper.myLooper();
}
@Override
public Poster createPoster(EventBus eventBus) {
return new HandlerPoster(eventBus, looper, 10);
}
}
AndroidHandlerMainThreadSupport的createPoster方法最终返回了一个HandlerPoster对象，并将主线程的looper传了进去。因此，可以确定，mainThreadPoster其实就是HandlerPoster类型的。我们进一步追踪代码。
public class HandlerPoster extends Handler implements Poster {
private final PendingPostQueue queue;
private final int maxMillisInsideHandleMessage;
private final EventBus eventBus;
private boolean handlerActive;
protected HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
super(looper);
this.eventBus = eventBus;
this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
queue = new PendingPostQueue();
}
...
}
HandlerPoster不仅实现了Poster接口，也继承了Handler类，覆写了其handleMessage方法。HandlerPoster将looper对象传递给了父类Handler，而该looper从前面分析可知是主线程looper，因此handleMessage方法最终也是运行在主线程的。HandlerPoster内部还封装了一个PendingPostQueue队列，该队列的元素是PendingPost，它封装了事件和订阅信息，并且持有指向下一个PendingPost对象的引用。PendingPost内部本身持有一个List<PendingPost>对象池，大小限制是10000，可以避免对象的频繁创建和回收，我们的应用也可以借鉴这种写法。
接下来看一下HandlerPost的enqueue实现
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);
if (!handlerActive) {
handlerActive = true;
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
}
}
}
enqueue的大致逻辑是这样的：
- 从PendingPost对象池获取一个PendingPost对象，该对象封装了订阅信息和事件
- 将PendingPost入队
- 如果handler没有处于激活态，则将handlerActive置位成true，并调用Handler的sendMessage发送消息
之后就需要在Handler中处理消息，即执行handleMessage。
@Override
public void handleMessage(Message msg) {
boolean rescheduled = false;
try {
long started = SystemClock.uptimeMillis();
while (true) {
PendingPost pendingPost = queue.poll();
if (pendingPost == null) {
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll();
if (pendingPost == null) {
handlerActive = false;
return;
}
}
}
eventBus.invokeSubscriber(pendingPost);
long timeInMethod = SystemClock.uptimeMillis() - started;
if (timeInMethod >= maxMillisInsideHandleMessage) {
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
rescheduled = true;
return;
}
}
} finally {
handlerActive = rescheduled;
}
}
这里在while循环里，首先从队列的头部取出PendingPost对象，如果为null，就直接跳出while循环。否则就向下调用EventBus的invokeSubscriber方法。执行完成后，需要判断单次执行handleMessage的时间是否超过预设定的最大值10ms，如果超过，则直接跳出循环，并将handlerActive置位成true。这样做，可以防止主线程因执行handleMessage过久而导致阻塞。
而invokeSubscriber方法的逻辑如下：
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);
}
}
首先是从PendingPost对象里面取出事件和订阅信息，并将PendingPost对象释放回对象池；之后直接调用invoke执行订阅方法。回溯上述流程，invoke方法其实是在handleMessage当中执行的，而handleMessage是执行在主线程的，因为其looper是mainLooper。到这里，我们终于知道了，当订阅方法的threadMode是MAIN或MAIN_ORDERED且订阅方法位于非UI线程时，EventBus是如何切换到UI线程的了。
三、BackgroundPoster
BackgroundPoster也是Poster的实现类，同时它也实现了Runnable接口。在它的内部同样封装了一个PendingPost队列，同时还有一个volatile类型de、用于判断线程池是否运行的布尔变量。
final class BackgroundPoster implements Runnable, Poster {
private final PendingPostQueue queue;
private final EventBus eventBus;
private volatile boolean executorRunning;
...
}
同样，我们来看一下它的enqueue方法。
public void enqueue(Subscription subscription, Object event) {
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost);
if (!executorRunning) {
executorRunning = true;
eventBus.getExecutorService().execute(this);
}
}
}
enqueue方法首先也是从PendingPost对象池中取出PendingPost对象并入队，之后调用EventBus的内置缓存线程池来执行任务，该任务就是BackgroundPoster自身，执行时会运行期内部的run方法。
@Override
public void run() {
try {
try {
while (true) {
PendingPost pendingPost = queue.poll(1000);
if (pendingPost == null) {
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll();
if (pendingPost == null) {
executorRunning = false;
return;
}
}
}
eventBus.invokeSubscriber(pendingPost);
}
} catch (InterruptedException e) {
eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
}
} finally {
executorRunning = false;
}
}
同样是在while循环里面，从队列中取出pendingPost，当pendingPost 为null时，需要将executorRunning置位成false并跳出循环，这样可以防止浪费CPU。当下次有新的事件时，再提交任务到线程池执行。如果pendingPost有值，则去调用invokeSubscriber方法，这部分逻辑和第二小节一致，不再赘述。只不过，此时的invoke方法是执行在后台线程中的。到这里，BackgroundPoster的切换操作也讲完了。
四、AsyncPoster
AsyncPoster也是Poster的实现类，同时也实现了Runnable接口。我们直接看它的enqueue和run方法。
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);
}
和BackgroundPoster的enqueue方法是不是非常相似？唯一不同的是，BackgroundPoster会先判断当前是否正在使用线程池处理任务，如果不是，才允许提交新的任务。这里其实可以看出，Background线程和Async线程的区别：Background线程会尽可能地在一个任务里有序地处理所有需要在Background线程模式下订阅的事件；而Async线程则是每个任务只处理一个事件，是无序的。因此，当需要在子线程执行订阅事件时，且订阅事件耗时较短或者需要保证顺序，笔者建议使用Background的ThreadMode；如果订阅事件耗时较长或不在意顺序，笔者建议使用Async的ThreadMode。
五、结束语
本章我们分析了EventBus最重要的线程切换相关的几个Poster，最后我们再总结一下不同线程模式的区别：
POSTING:事件从哪个线程发出，就在哪个线程执行订阅方法
MAIN：事件无论从哪个线程发出，都在主线程执行订阅方法
MAIN_ORDERED：事件无论从哪个线程发出，都在主线程执行订阅方法
BACKGROUND:事件从主线程发出，则在后台线程执行订阅方法；事件从子线程发出，则直接在该线程执行订阅方法
ASYNC:事件无论从哪个线程发出，都开启新的子线程处理
网友评论