【EventBus】EventBus 源码解析 ( 事件发送 | 线程池中执行订阅方法 )

一、EventBus 中主线程支持类

从 Subscription subscription 参数中 , 获取订阅方法的线程模式 , 根据 【EventBus】Subscribe 注解分析 ( Subscribe 注解属性 | threadMode 线程模型 | POSTING | MAIN | MAIN_ORDERED | ASYNC) 博客的运行规则 , 执行线程 ;

如果订阅方法的线程模式被设置为 ASYNC , 则不管在哪个线程中发布消息 , 都会将事件放入队列 , 通过线程池执行该事件 ;

public class EventBus {
    private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
    	// 获取该 订阅方法 的线程模式 
        switch (subscription.subscriberMethod.threadMode) {
            case ASYNC:
                asyncPoster.enqueue(subscription, event);
                break;
            default:
                throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
        }
    }
}

二、EventBus 中 AsyncPoster 分析

AsyncPoster 分析 : 在 EventBus 中 , 定义了 AsyncPoster asyncPoster 成员变量 , 在构造函数中进行了初始化操作 ;

public class EventBus {
    private final AsyncPoster asyncPoster;

    EventBus(EventBusBuilder builder) {
        asyncPoster = new AsyncPoster(this);
    }
}

三、AsyncPoster 线程池 Runnable 任务类

AsyncPoster 实现了 Runnable 接口 , 在 run 方法中 , 调用 eventBus.invokeSubscriber(pendingPost) 执行订阅方法 ;

将该 Runnable 实现类 , 直接传递给线程池 , 即可执行 ;

/**
 * Posts events in background.
 * 
 * @author Markus
 */
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 = PendingPost.obtainPendingPost(subscription, event);
        // 将 订阅者 和 事件 加入到 PendingPost 链表中 
        queue.enqueue(pendingPost);
        // 启动线程池执行 AsyncPoster 任务
        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);
    }

}