标签：sub illegal udp ber usb 4行 interrupt enqueue npe
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提供的线程。
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队列中有元素却没有得到及时处理。
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保证的是串行化，但并不是使用同一线程实现的。
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请求。
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修饰。
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，并提供静态对象池，用于对象复用。
标签：sub illegal udp ber usb 4行 interrupt enqueue npe
原文地址：http://www.cnblogs.com/moderate-fish/p/7693870.html
