EventBus3.0源码学习 threadMode

Posted 2020-10-11 荆棘鸟

线程模式

• POSTING：该模式不需要线程切换，为默认模式。在post线程直接调用事件订阅函数。
• MAIN：若post线程是主线程，则与POSTING模式一样；否则，由mainThreadPoster进行处理，从主线程调用事件订阅函数。
• BACKGROUND：若post线程不是主线程，则与POSTING一样；否则，由backgroundPoster进行处理，从background线程调用事件订阅函数。
• ASYNC：该模式总是需要进行线程切换，由asyncPoster进行处理。

EventBus中与线程相关的成员变量如下：

public class EventBus {
    private final HandlerPoster mainThreadPoster; // MAIN
    private final BackgroundPoster backgroundPoster; // BACKGROUND
    private final AsyncPoster asyncPoster; // ASYNC
    private final ExecutorService executorService; // Executors.newCachedThreadPool()

    EventBus(EventBusBuilder builder) {
        mainThreadPoster = new HandlerPoster(this, Looper.getMainLooper(), 10);
        backgroundPoster = new BackgroundPoster(this);
        asyncPoster = new AsyncPoster(this);
        executorService = builder.executorService;
    }
}

mainThreadPoster使用mainLooper，而backgroundPoster、syncPoster都是使用exectuorService提供的线程。

HandlerPoster

 1 final class HandlerPoster extends Handler {
 2 
 3     private final PendingPostQueue queue;
 4     private final int maxMillisInsideHandleMessage;
 5     private final EventBus eventBus;
 6     private boolean handlerActive;
 7 
 8     HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
 9         super(looper);
10         this.eventBus = eventBus;
11         this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
12         queue = new PendingPostQueue();
13     }
14 
15     void enqueue(Subscription subscription, Object event) {
16         PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
17         synchronized (this) {
18             queue.enqueue(pendingPost);
19             if (!handlerActive) {
20                 handlerActive = true;
21                 if (!sendMessage(obtainMessage())) {
22                     throw new EventBusException("Could not send handler message");
23                 }
24             }
25         }
26     }
27 
28     @Override
29     public void handleMessage(Message msg) {
30         boolean rescheduled = false;
31         try {
32             long started = SystemClock.uptimeMillis();
33             while (true) {
34                 PendingPost pendingPost = queue.poll();
35                 if (pendingPost == null) {
36                     synchronized (this) {
37                         // Check again, this time in synchronized
38                         pendingPost = queue.poll();
39                         if (pendingPost == null) {
40                             handlerActive = false;
41                             return;
42                         }
43                     }
44                 }
45                 eventBus.invokeSubscriber(pendingPost);
46                 long timeInMethod = SystemClock.uptimeMillis() - started;
47                 if (timeInMethod >= maxMillisInsideHandleMessage) {
48                     if (!sendMessage(obtainMessage())) {
49                         throw new EventBusException("Could not send handler message");
50                     }
51                     rescheduled = true;
52                     return;
53                 }
54             }
55         } finally {
56             handlerActive = rescheduled;
57         }
58     }
59 }

第15-26行，enqueue操作将subscription和event封装成一个pendingPost对象，将其放入queue队列中。若handlerActive为false，表明主线程没有正在处理queue中的消息，则调用sendMessage(obtainMessage())通知Looper线程处理。

HandlerPoster继承Handler，enqueue操作调用sendMessage，会导致handleMessage回调从Looper线程中触发。第33-54行，handleMessage中使用一个循环处理queue队列中的消息。若queue队列为空，则退出循环；若当前处理事件总耗时超过了maxMillisInsideHandleMessage，则调用sendMessage并退出循环(此时handlerActive仍为true，因为Looper线程会在处理完handleMessage运行期间积压的事务之后再次触发handleMessage回调)。

第36-43行的synchronized块和第17-25行的synchronized块，共同保证handlerActive的值被正确设置。不过，第56行没有在synchronized块中，这可能导致sendMessage被重复调用，但并不会出现queue队列中有元素却没有得到及时处理。

BackgroundPoster

 1 final class BackgroundPoster implements Runnable {
 2 
 3     private final PendingPostQueue queue;
 4     private final EventBus eventBus;
 5 
 6     private volatile boolean executorRunning;
 7 
 8     BackgroundPoster(EventBus eventBus) {
 9         this.eventBus = eventBus;
10         queue = new PendingPostQueue();
11     }
12 
13     public void enqueue(Subscription subscription, Object event) {
14         PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
15         synchronized (this) {
16             queue.enqueue(pendingPost);
17             if (!executorRunning) {
18                 executorRunning = true;
19                 eventBus.getExecutorService().execute(this);
20             }
21         }
22     }
23 
24     @Override
25     public void run() {
26         try {
27             try {
28                 while (true) {
29                     PendingPost pendingPost = queue.poll(1000);
30                     if (pendingPost == null) {
31                         synchronized (this) {
32                             // Check again, this time in synchronized
33                             pendingPost = queue.poll();
34                             if (pendingPost == null) {
35                                 executorRunning = false;
36                                 return;
37                             }
38                         }
39                     }
40                     eventBus.invokeSubscriber(pendingPost);
41                 }
42             } catch (InterruptedException e) {
43                 Log.w("Event", Thread.currentThread().getName() + " was interruppted", e);
44             }
45         } finally {
46             executorRunning = false;
47         }
48     }
49 
50 }

第13-22行，enqueue操作将subscription和event封装成一个pendingPost对象，将其放入queue队列中。若executorRunning为false，表明background线程没有正在处理queue队列中的消息，则调用eventBus.getExecutorService().execute(this)执行backgroundPoster。

BackgroundPoster继承Runnable，run中使用一个循环处理queue队列中的消息。第29行，若queue队列为空时，poll(1000)会等待1000毫秒，若超时后还没取到消息，则进入synchronized块再poll一次，若此时若仍返回空，则退出循环。

第31-38行的synchronized块和第15-21行的synchronized块，共同保证executorRunning的值被正确设置。不过，第46行没有在synchronized块中，这可能导致backgroundPost同时运行在两个线程中，这跟backgroudPoster的定义是乎并不一致。

从代码实现来看，BackgroundPoster保证的是串行化，但并不是使用同一线程实现的。

AsyncPoster

 1 class AsyncPoster implements Runnable {
 2 
 3     private final PendingPostQueue queue;
 4     private final EventBus eventBus;
 5 
 6     AsyncPoster(EventBus eventBus) {
 7         this.eventBus = eventBus;
 8         queue = new PendingPostQueue();
 9     }
10 
11     public void enqueue(Subscription subscription, Object event) {
12         PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
13         queue.enqueue(pendingPost);
14         eventBus.getExecutorService().execute(this);
15     }
16 
17     @Override
18     public void run() {
19         PendingPost pendingPost = queue.poll();
20         if(pendingPost == null) {
21             throw new IllegalStateException("No pending post available");
22         }
23         eventBus.invokeSubscriber(pendingPost);
24     }
25 
26 }

AsyncPoster继承Runnable，enqueue操作将subscription和event封装成一个pendingPost对象，将其放入queue队列中。每次enqueue操作都将调用eventBus.getExecutorService().execute(this)，从线程池获取一个线程执行asyncPoster，处理queue队列中的消息。每次线程只执行一个pendingPost请求。

PendingPostQueue结构

 1 final class PendingPostQueue {
 2     private PendingPost head;
 3     private PendingPost tail;
 4 
 5     synchronized void enqueue(PendingPost pendingPost) {
 6         if (pendingPost == null) {
 7             throw new NullPointerException("null cannot be enqueued");
 8         }
 9         if (tail != null) {
10             tail.next = pendingPost;
11             tail = pendingPost;
12         } else if (head == null) {
13             head = tail = pendingPost;
14         } else {
15             throw new IllegalStateException("Head present, but no tail");
16         }
17         notifyAll();
18     }
19 
20     synchronized PendingPost poll() {
21         PendingPost pendingPost = head;
22         if (head != null) {
23             head = head.next;
24             if (head == null) {
25                 tail = null;
26             }
27         }
28         return pendingPost;
29     }
30 
31     synchronized PendingPost poll(int maxMillisToWait) throws InterruptedException {
32         if (head == null) {
33             wait(maxMillisToWait);
34         }
35         return poll();
36     }
37 
38 }

PendingPostQueue底层使用head和tail维护一个PendingPost队列。enqueue和poll操作都由synchronized修饰。

PendingPost

 1 final class PendingPost {
 2     private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();
 3 
 4     Object event;
 5     Subscription subscription;
 6     PendingPost next;
 7 
 8     private PendingPost(Object event, Subscription subscription) {
 9         this.event = event;
10         this.subscription = subscription;
11     }
12 
13     static PendingPost obtainPendingPost(Subscription subscription, Object event) {
14         synchronized (pendingPostPool) {
15             int size = pendingPostPool.size();
16             if (size > 0) {
17                 PendingPost pendingPost = pendingPostPool.remove(size - 1);
18                 pendingPost.event = event;
19                 pendingPost.subscription = subscription;
20                 pendingPost.next = null;
21                 return pendingPost;
22             }
23         }
24         return new PendingPost(event, subscription);
25     }
26 
27     static void releasePendingPost(PendingPost pendingPost) {
28         pendingPost.event = null;
29         pendingPost.subscription = null;
30         pendingPost.next = null;
31         synchronized (pendingPostPool) {
32             // Don‘t let the pool grow indefinitely
33             if (pendingPostPool.size() < 10000) {
34                 pendingPostPool.add(pendingPost);
35             }
36         }
37     }
38 
39 }

PendingPost封装了subscription和event，并提供静态对象池，用于对象复用。