EventBus源码分析

作者: Lemon666 | 来源:发表于2021-02-24 18:42 被阅读0次

简介

源码基于 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进行注销，清空之前订阅的方法

EventBus源码分析

简介

使用步骤

原理分析（图片）

原理分析（源代码）

总结

准备阶段

发送阶段

注销阶段

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读