EventBus源码分析

简介

源码基于 org.greenrobot:eventbus:3.2.0
EventBus是Android和Java的发布/订阅事件总线。

使用步骤

自定义消息

public static class MessageEvent { 
    //可以根据业务需求添加所需的字段
    public int code;//定义一个业务code，可区分不同消息
}

声明并注解订阅方法，可以指定运行线程

@Subscribe(threadMode = ThreadMode.MAIN)  //ThreadMode.MAIN指定在主线程运行
public void onMessageEvent(MessageEvent event) {
    //获取到MessageEvent消息体后，可在这进行业务处理
};

在Activity生命周期注册和反注册EventBus

 @Override
 public void onStart() {
     super.onStart();
    //注册
     EventBus.getDefault().register(this);
 }

 @Override
 public void onStop() {
     super.onStop();
    //反注册
     EventBus.getDefault().unregister(this);
 }

准备完毕后，就可以开始发送一个自定义消息MessageEvent

//可以在主线程也可在非主线程进行发送
EventBus.getDefault().post(new MessageEvent());

以上就是EventBus的简单使用，下面来分析一下EventBus，自定义消息如何发送到指定方法上去并且在是如何指定运行线程。

原理分析（图片）

准备分析前，先通过图来简单说明一下大致流程

刚开始，我们未订阅任何方法和未发送任何消息事件

image.png

假设在EventActivity订阅一个onMessageEvent方法，并且该方法接收MessageEvent消息事件，刚开始相互没有关联

image.png
当Activity调用onStart时候，EventBus进行了注册，这个时候就开始关联起来，

 @Override
 public void onStart() {
     super.onStart();
    //注册
     EventBus.getDefault().register(this);
 }

image.png

相应的，当反注册的时候，将移除对应关系

发送消息事件

image.png

原理分析（源代码）

EventBus使用了单利模式，所以全局只有一个EventBus实例，使用DLC

    public static EventBus getDefault() {
        EventBus instance = defaultInstance;
        if (instance == null) {
            synchronized (EventBus.class) {
                instance = EventBus.defaultInstance;
                if (instance == null) {
                    instance = EventBus.defaultInstance = new EventBus();
                }
            }
        }
        return instance;
    }

在开始分析前，先了解EventBus里面2个字段，都是Map+List的数据格式
第一个是存放一个消息类型对应的全部订阅方法
第二个是存放和注册对象有关联的消息类型

  //根据消息事件类型进行存储所有订阅方法
  private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;
  //根据注册的实例存储当前对象订阅方法所有接收消息事件
  private final Map<Object, List<Class<?>>> typesBySubscriber;

以我们前面使用步骤代码为例，假设当前代码在EventActivity

class EventActivity{
    //消息
    public static class MessageEvent { }

    //订阅方法
    @Subscribe(threadMode = ThreadMode.MAIN)  //ThreadMode.MAIN指定在主线程运行
    public void onMessageEvent(MessageEvent event) { };

    @Override
    public void onStart() {
      super.onStart();
       EventBus.getDefault().register(this);
    }

    @Override
    public void onStop() {
       super.onStop();
       EventBus.getDefault().unregister(this);
    }
}

通过注册进行关联

public void register(Object subscriber) {
    //获取注册对象的类类型，这里是EventActivity.class
    Class<?> subscriberClass = subscriber.getClass();
    //根据注解获取EventActivity所有订阅方法，并且封装为SubscriberMethod对象，因为Activity可以订阅多个方法，
    //所以最终返回的是一个数组
    List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
    synchronized (this) {
        //遍历所有订阅方法，将订阅方法存放到指定的位置
        for (SubscriberMethod subscriberMethod : subscriberMethods) {
            //订阅
            subscribe(subscriber, subscriberMethod);
        }
    }
}

图示返回3个SubscriberMethod，比较好表示为数组，实际应该返回1个

接下来继续看如何进行订阅

//订阅
//Object subscriber这里为EventActivty对象
private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
    //获取订阅方法接收的消息类型，这里为MessageEvent.class
    Class<?> eventType = subscriberMethod.eventType;
    //将EventActivity对象和订阅方法组合为Subscription
    Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
    //根据消息类型获取该消息存放订阅方法的数组
    CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
    if (subscriptions == null) {
        //为空，也就是该消息第一次订阅，则创建一个新的数组，并且添加到map中
        subscriptions = new CopyOnWriteArrayList<>();
        subscriptionsByEventType.put(eventType, subscriptions);
    } else {
        //同一个订阅方法不能重复订阅
        if (subscriptions.contains(newSubscription)) {
            throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
                    + eventType);
        }
    }
    //这里主要是根据优先级将订阅方法插入指定位置
    int size = subscriptions.size();
    for (int i = 0; i <= size; i++) {
        if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
            //将订阅方法插入数组，这样就完成一次订阅方法的添加
            subscriptions.add(i, newSubscription);
            break;
        }
    }
    //获取该对象（EventActivity）全部订阅方法接收的消息事件
    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
    if (subscribedEvents == null) {
        //第一次创建一个并且设置到map
        subscribedEvents = new ArrayList<>();
        typesBySubscriber.put(subscriber, subscribedEvents);
    }
    //添加消息类型
    subscribedEvents.add(eventType);
    ...
}

image.png

发送一个MessageEvent

public void post(Object event) {
    //获取当前线程的PostingThreadState
    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 postSingleEvent(Object event, PostingThreadState postingState) throws Error {
    //获取当前消息的类类型，这里是MessagEvent.class
    Class<?> eventClass = event.getClass();
    //是否找到订阅方法
    boolean subscriptionFound = false;
    if (eventInheritance) {
        //是否查询MessagEvent.class的父类类型，也就是父类型的消息订阅的方法也要进行发送
        List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
        int countTypes = eventTypes.size();
        for (int h = 0; h < countTypes; h++) {
            Class<?> clazz = eventTypes.get(h);
            //发送消息
            subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
        }
    } else {
        //发送消息
        subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
    }
    if (!subscriptionFound) {
        //未找到订阅方法进行相应处理
        if (logNoSubscriberMessages) {
            logger.log(Level.FINE, "No subscribers registered for event " + eventClass);
        }
        if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
                eventClass != SubscriberExceptionEvent.class) {
            post(new NoSubscriberEvent(this, event));
        }
    }
}

private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
    //定义一个订阅方法的数组
    CopyOnWriteArrayList<Subscription> subscriptions;
    synchronized (this) {
        //根据事件类型获取全部订阅方法的数组
        subscriptions = subscriptionsByEventType.get(eventClass);
    }
    //存在订阅方法
    if (subscriptions != null && !subscriptions.isEmpty()) {
        //遍历所有订阅方法
        for (Subscription subscription : subscriptions) {
            //标记当前发送的事件和订阅方法
            postingState.event = event;
            postingState.subscription = subscription;
            boolean aborted;
            try {
                //发送事件
                postToSubscription(subscription, event, postingState.isMainThread);
                aborted = postingState.canceled;
            } finally {
                //重置
                postingState.event = null;
                postingState.subscription = null;
                postingState.canceled = false;
            }
            if (aborted) {
                //如果中途中断则跳出循环
                break;
            }
        }
        return true;
    }
    return false;
}

private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
    switch (subscription.subscriberMethod.threadMode) {
        case POSTING://在哪个线程发送消息，就在哪个线程进行接收处理
            invokeSubscriber(subscription, event);
            break;
        case MAIN://主线程接收处理
            if (isMainThread) {
                //当前是主线程，直接发送
                invokeSubscriber(subscription, event);
            } else {
                //当前非主线程，则通过mainThreadPoster发送到主线程
                mainThreadPoster.enqueue(subscription, event);
            }
            break;
        case MAIN_ORDERED://主线程按顺序接收处理
            if (mainThreadPoster != null) {
                mainThreadPoster.enqueue(subscription, event);
            } else {
                // temporary: technically not correct as poster not decoupled from subscriber
                invokeSubscriber(subscription, event);
            }
            break;
        case BACKGROUND:
            if (isMainThread) {
                backgroundPoster.enqueue(subscription, event);
            } else {
                invokeSubscriber(subscription, event);
            }
            break;
        case ASYNC:
            asyncPoster.enqueue(subscription, event);
            break;
        default:
            throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
    }
}

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);
    }
}

子线程切换成主线程

//主线程HandlerPoster继承自Handler
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();
    }

    public void enqueue(Subscription subscription, Object event) {
        //消息入队
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            queue.enqueue(pendingPost);
            if (!handlerActive) {
                handlerActive = true;
                //发送消息到主线程的Looper
                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 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;
        }
    }
}

主线程切换成子线程

final class BackgroundPoster implements Runnable, Poster {
    //队列
    private final PendingPostQueue queue;
    private final EventBus eventBus;

    private volatile boolean executorRunning;

    BackgroundPoster(EventBus eventBus) {
        this.eventBus = eventBus;
        queue = new PendingPostQueue();
    }

    public void enqueue(Subscription subscription, Object event) {
        //消息入队
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            queue.enqueue(pendingPost);
            if (!executorRunning) {
                executorRunning = true;
                //通过EventBus的线程池执行
                //这样就完成一次主线程到子线程的切换
                eventBus.getExecutorService().execute(this);
            }
        }
    }

    @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;
        }
    }
}

最后看下反注册如何处理

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());
    }
}

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--;
            }
        }
    }
}

这样就反注册移除全部订阅方法

总结

准备阶段

创建消息事件对象
添加一个订阅方法并且添加注解
调用EventBus进行注册

发送阶段

通过EventBus.post发送前面创建的消息事件对象

注销阶段

调用EventBus进行注销，清空之前订阅的方法

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 204,732评论 6赞 478
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 87,496评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 151,264评论 0赞 338
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,807评论 1赞 277
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,806评论 5赞 368
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,675评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,029评论 3赞 399
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,683评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 41,704评论 1赞 299
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,666评论 2赞 321
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,773评论 1赞 332
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,413评论 4赞 321
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,016评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,978评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,204评论 1赞 260
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,083评论 2赞 350
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,503评论 2赞 343

EventBus源码分析

简介

使用步骤

原理分析（图片）

原理分析（源代码）

总结

准备阶段

发送阶段

注销阶段

推荐阅读更多精彩内容