上面这个发送方法中用到了3个post事件，下面我们把这几个post事件分析一下：
主线程的调度：
- MainThreadSupport：
/**
* Interface to the "main" thread, which can be whatever you like. Typically on Android, Android's main thread is used.
* 接口到“主”线程，可以是任何你喜欢的。通常在Android上使用Android的主线程。
*/
public interface MainThreadSupport {
boolean isMainThread();
Poster createPoster(EventBus eventBus);
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);
}
}
}
从上方法中可以看出最终创建的poster是HandlerPoster；
- HandlerPoster
public class HandlerPoster extends Handler implements Poster {
private final PendingPostQueue queue; // 存放待执行的 Post Events 的事件队列
private final int maxMillisInsideHandleMessage; // post 事件在 handlerMessage 中执行的最大的时间值，超过这个时间值会抛出异常
private final EventBus eventBus;
private boolean handlerActive; // 标识 handler 是否被运行起来了
protected HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
super(looper);
this.eventBus = eventBus;
this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
queue = new PendingPostQueue();
}
public void enqueue(Subscription subscription, Object event) {
// 从 pendingPostPool 的 ArrayList 缓存池中获取 PendingPost 添加到 PendingPostQueue 队列中，并将该 PendingPost 事件发送到 Handler 中处理
PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
synchronized (this) {
queue.enqueue(pendingPost); // 把 pendingPost 添加到队列中
if (!handlerActive) { // 标记 Handler 为活跃状态
handlerActive = true;
if (!sendMessage(obtainMessage())) {
throw new EventBusException("Could not send handler message");
}
}
}
}
@Override
public void handleMessage(Message msg) {
boolean rescheduled = false;
try {
long started = SystemClock.uptimeMillis();
while (true) { // 死循环，不断从 PendingPost 队列中取出 post 事件执行
PendingPost pendingPost = queue.poll();
if (pendingPost == null) { // 如果为 null，表示队列中没有 post 事件，此时标记 Handler 关闭，并退出 while 循环
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll();
if (pendingPost == null) {
handlerActive = false; // 标记 Handler 为非活跃状态
return;
}
}
}
// 获取到 post 之后由 eventBus 通过该 post 查找相应的 Subscriber 处理事件
eventBus.invokeSubscriber(pendingPost);
// 计算每个事件在 handleMessage 中执行的时间
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;
}
}
}
HandlerPoster 其实就是 Handler 的实现，内部维护了一个 PendingPostQueue 的消息队列，在 enqueue(Subscription subscription, Object event) 方法中不断从 pendingPostPool 的 ArrayList 缓存池中获取 PendingPost 添加到 PendingPostQueue 队列中，并将该 PendingPost 事件发送到 Handler 中处理。
在 handleMessage 中，通过一个 while 死循环，不断从 PendingPostQueue 中取出 PendingPost 出来执行，获取到 post 之后由 eventBus 通过该 post 查找相应的 Subscriber 处理事件。
while 退出的条件有两个
1.获取到的 PendingPost 为 null，即是 PendingPostQueue 已经没有消息可处理。
2.每个 PendingPost 在 Handler 中执行的时间超过了最大的执行时间。2.BackgroundPoster
/**
* Posts events in background.
* 在后台发布事件
* @author Markus
*/
final class BackgroundPoster implements Runnable, Poster {
.....
@Override
public void run() {
try {
try {
// 不断循环从 PendingPostQueue 取出 pendingPost 到 eventBus 执行
while (true) {
// 在 1000 毫秒内从 PendingPostQueue 中获取 pendingPost
PendingPost pendingPost = queue.poll(1000);
// 双重校验锁判断 pendingPost 是否为 null
if (pendingPost == null) {
synchronized (this) {
// Check again, this time in synchronized
pendingPost = queue.poll(); // 再次尝试获取 pendingPost
if (pendingPost == null) {
executorRunning = false;
return;
}
}
}
// 将 pendingPost 通过 EventBus 分发出去
// 这里会将 PendingPostQueue 中【所有】的 pendingPost 都会分发，这里区别于 AsyncPoster
eventBus.invokeSubscriber(pendingPost);
}
} catch (InterruptedException e) {
eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
}
} finally {
executorRunning = false;
}
}
}
BackgroundPoster实现的是Runnable，我们看它的run方法，不断从PendingPostQueue 中取出 pendingPost 到 EventBus 中分发，这里注意外部是 while() 死循环，意味着 PendingPostQueue 中所有的 pendingPost 都将分发出去。
原因就是：BackGroundThread后台线程，这是唯一的，所以发送到这个线程中的事件需要放到一个队列中，订阅的时候循环遍历全部发出去。
- 3.AsyncPoster
这个poster和BackgroundPoster都是实现的Runnable，但是他们的run方法实现却不一样，只是取出一个分发出去。
原因：ASYNC:这个线程，不管发布事件是在哪个线程中，订阅的时候都会在一个空线程中，所以线程都相互独立不会出现卡顿。
class AsyncPoster implements Runnable, Poster {
.....
@Override
public void run() {
PendingPost pendingPost = queue.poll();
if(pendingPost == null) {
throw new IllegalStateException("No pending post available");
}
// 将 pendingPost 通过 EventBus 分发出去
eventBus.invokeSubscriber(pendingPost);
}
}
3个poster中用到的队列
PendingPostQueue:
final class PendingPostQueue {
private PendingPost head; // 头
private PendingPost tail; // 尾
// 入队
synchronized void enqueue(PendingPost pendingPost) {
if (pendingPost == null) {
throw new NullPointerException("null cannot be enqueued");
}
if (tail != null) { // 有尾
tail.next = pendingPost;
tail = pendingPost;
} else if (head == null) { // 队列的头部和尾部都为空
head = tail = pendingPost; // 队列为空，添加第一个元素进队列
} else {
throw new IllegalStateException("Head present, but no tail"); // 有头无尾
}
notifyAll(); // 唤醒等候的全部线程
}
synchronized PendingPost poll() {
PendingPost pendingPost = head;
if (head != null) {
head = head.next;
if (head == null) {
tail = null;
}
}
return pendingPost;
}
synchronized PendingPost poll(int maxMillisToWait) throws InterruptedException {
if (head == null) {
wait(maxMillisToWait);
}
return poll();
}
}
- PendingPost:
final class PendingPost {
// 通过ArrayList来实现PendingPost的添加和删除
private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();
Object event;
Subscription subscription;
PendingPost next;
private PendingPost(Object event, Subscription subscription) {
this.event = event;
this.subscription = subscription;
}
// 获取 PendingPost
static PendingPost obtainPendingPost(Subscription subscription, Object event) {
synchronized (pendingPostPool) {
int size = pendingPostPool.size();
if (size > 0) {
PendingPost pendingPost = pendingPostPool.remove(size - 1); // 返回一个PendingPost，就把该对象从缓存池中移除出去
pendingPost.event = event;
pendingPost.subscription = subscription;
pendingPost.next = null;
return pendingPost;
}
}
return new PendingPost(event, subscription); // pendingPostPool缓存池中没有了就新创建一个
}
// 释放 PendingPost
static void releasePendingPost(PendingPost pendingPost) {
pendingPost.event = null;
pendingPost.subscription = null;
pendingPost.next = null;
synchronized (pendingPostPool) {
// Don't let the pool grow indefinitely
if (pendingPostPool.size() < 10000) { // 缓存池中最多存放一万条数据
pendingPostPool.add(pendingPost);
}
}
}
}
5.注销
EventBus.getDefault().unregister(this);
- unregister方法：
public synchronized void unregister(Object subscriber) {
List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
if (subscribedTypes != null) {
for (Class<?> eventType : subscribedTypes) {
unsubscribeByEventType(subscriber, eventType);
}
typesBySubscriber.remove(subscriber);
} else {
logger.log(Level.WARNING, "Subscriber to unregister was not registered before: " + subscriber.getClass());
}
}
注册过程我们就知道 typesBySubscriber 是保存订阅者的所有订阅事件类型的一个 Map，这里根据订阅者拿到订阅事件类型 List，然后逐个取消订阅，最后 typesBySubscriber 移除该订阅者。
- unsubscribeByEventType方法：
/**
* Only updates subscriptionsByEventType, not typesBySubscriber! Caller must update typesBySubscriber.
* 只更新subscriptionByEventType，不更新typesbysubscriber！调用方必须更新typesbysubscriber。
*/
private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
if (subscriptions != null) {
int size = subscriptions.size();
for (int i = 0; i < size; i++) {
Subscription subscription = subscriptions.get(i);
if (subscription.subscriber == subscriber) {
subscription.active = false;
subscriptions.remove(i);
i--;
size--;
}
}
}
}
subscriptionsByEventType 是存储事件类型对应订阅信息的 Map，代码逻辑非常清晰，找出某事件类型的订阅信息 List，遍历订阅信息，将要取消订阅的订阅者和订阅信息封装的订阅者比对，如果是同一个，则说明该订阅信息是将要失效的，于是将该订阅信息移除。
