首页 > 代码库 > 聊一聊Android的消息机制

聊一聊Android的消息机制

聊一聊Android的消息机制

 

侯 亮

1概述

在Android平台上,主要用到两种通信机制,即Binder机制和消息机制,前者用于跨进程通信,后者用于进程内部通信。

从技术实现上来说,消息机制还是比较简单的。从大的方面讲,不光是Android平台,各种平台的消息机制的原理基本上都是相近的,其中用到的主要概念大概有:
1)消息发送者;
2)消息队列;
3)消息处理循环。
示意图如下:

技术分享

图中表达的基本意思是,消息发送者通过某种方式,将消息发送到某个消息队列里,同时还有一个消息处理循环,不断从消息队列里摘取消息,并进一步解析处理。

 在Android平台上,把上图的右边部分包装成了一个Looper类,这个类的内部具有对应的消息队列(MessageQueue  mQueue)和loop函数。
 
但是Looper只是个简单的类而已,它虽然提供了循环处理方面的成员函数loop(),却不能自己凭空地运行起来,而只能寄身于某个真实的线程。而且,每个线程最多只能运作一个Looper对象,这一点应该很容易理解。

Android平台上另一个关键类是Handler。当消息循环在其寄身的线程里正式运作后,外界就是通过Handler向消息循环发出事件的。我们再画一张示意图如下:

技术分享

当然,系统也允许多个Handler向同一个消息队列发送消息:

技术分享

整个消息机制的轮廓也就是这些啦,下面我们来详细阐述。

2先说一下Looper部分

Looper类的定义截选如下:
【frameworks/base/core/java/android/os/Looper.java】

[java] view plain copy
 
  1. public final class Looper {  
  2.     private static final String TAG = "Looper";  
  3.   
  4.     // sThreadLocal.get() will return null unless you‘ve called prepare().  
  5.     static final ThreadLocal<Looper> sThreadLocal = new ThreadLocal<Looper>();  
  6.     private static Looper sMainLooper;  // guarded by Looper.class  
  7.   
  8.     final MessageQueue mQueue;  
  9.     final Thread mThread;  
  10.   
  11.     private Printer mLogging;  
  12.     . . . . . .  
  13.     . . . . . .  

 

当一个线程运行到某处,准备运作一个Looper时,它必须先调用Looper类的静态函数prepare(),做一些准备工作。说穿了就是创建一个Looper对象,并把它设置进线程的本地存储区(TLS)里。然后线程才能继续调用Looper类的另一个静态函数loop(),从而建立起消息处理循环。示意图如下:
技术分享

prepare()函数的代码如下:

[java] view plain copy
 
  1. public static void prepare()   
  2. {  
  3.     prepare(true);  
  4. }  
  5.   
  6. private static void prepare(boolean quitAllowed)   
  7. {  
  8.     if (sThreadLocal.get() != null) {  
  9.         throw new RuntimeException("Only one Looper may be created per thread");  
  10.     }  
  11.     sThreadLocal.set(new Looper(quitAllowed));  // 创建Looper对象,并设置进TLS  
  12. }  

可以看到,sThreadLocal.set()一句所完成的工作,正是把新创建的Looper对象设置进线程本地存储区里。在Looper.prepare()之后,线程的主运作函数就可以调用Looper.loop()了。

 

为了便于大家理解,我们多说两句关于sThreadLocal的细节,这会牵扯一点儿本地存储的技术。简单地说,每个线程对象内部会记录一张逻辑上的key-value表,当然,这张表在具体实现时不一定会被实现成HashMap,以我们目前的代码来说,它被记录成一个数组,其中每两个数组项作为一个key-value单元。反正大家从逻辑上理解概念即可,不必拘泥于具体实现。很明显,一个线程内部是可以记录多个本地存储单元的,我们关心的sThreadLocal只是其中一个本地存储单元的key而已。

当我们在不同Thread里调用Looper.prepare()时,其实是向Thread对应的那张表里添加一个key-value项,其中的key部分,指向的是同一个对象,即Looper.sThreadLocal静态对象,而value部分,则彼此不同,我们可以画出如下示意图:

技术分享

看到了吧,不同Thread会对应不同Object[]数组,该数组以每2个元素为一个key-value对。请注意不同Thread虽然使用同一个静态对象作为key值,最终却会对应不同的Looper对象,这一点系统是不会弄错的。

为了由浅入深地阐述问题,我们暂时先不看Looper.loop()内部的代码,这个后文还会再讲。现在我们接着说说Handler。

3接着说一下Handler部分

一般而言,运作Looper的线程会负责构造自己的Handler对象,当然,其他线程也可以针对某个Looper构造Handler对象。

Handler对象在构造时,不但会把Looper对象记录在它内部的mLooper成员变量中,还会把Looper对象的消息队列也一并记录,代码截选如下:

[java] view plain copy
 
  1. public Handler(Callback callback, boolean async)   
  2. {  
  3.     . . . . . .  
  4.     mLooper = Looper.myLooper();   // 记录下Looper对象  
  5.     . . . . . .  
  6.     mQueue = mLooper.mQueue;        // 也记录下Looper对象的消息队列  
  7.     mCallback = callback;  
  8.     mAsynchronous = async;  
  9. }  

我们也可以直接传入Looper对象,此时可以使用另一个构造函数:

[java] view plain copy
 
  1. public Handler(Looper looper, Callback callback, boolean async)   
  2. {  
  3.     mLooper = looper;                // 记录下Looper对象  
  4.     mQueue = looper.mQueue;         // 也记录下Looper对象的消息队列  
  5.     mCallback = callback;  
  6.     mAsynchronous = async;  
  7. }  

以后,每当线程需要向消息队列发送消息时,只需调用Handler对象的sendMessage()等成员函数就可以了。

简单说来,只要一个线程可以获取另一个目标线程的某个Handler对象,它就具有了向目标线程发送消息的能力。不过,也只是发送消息而已,消息的真正处理却是在目标线程的消息循环里完成的。

前文已经说过,在Looper准备停当后,我们的线程会调用Looper.loop(),从而进入真正的循环机制。loop()函数的代码流程非常简单,只不过是在一个for循环里不停从消息队列中摘取消息,而后调用msg.target.dispatchMessage()对消息进行派发处理而已。

这么看来,msg.target域就显得比较重要了,说穿了,这个域记录的其实就是当初向消息队列发送消息的那个handler啦。当我们调用handler的send函数时,最终基本上都会走到sendMessageAtTime(),其代码如下:
【frameworks/base/core/java/android/os/Handler.java】

[java] view plain copy
 
  1. public boolean sendMessageAtTime(Message msg, long uptimeMillis)   
  2. {  
  3.     MessageQueue queue = mQueue;  
  4.     if (queue == null) {  
  5.         RuntimeException e = new RuntimeException(  
  6.                 this + " sendMessageAtTime() called with no mQueue");  
  7.         Log.w("Looper", e.getMessage(), e);  
  8.         return false;  
  9.     }  
  10.     return enqueueMessage(queue, msg, uptimeMillis);  
  11. }  

 

[java] view plain copy
 
  1. private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis)   
  2. {  
  3.     // 注意这一句,消息的target就是handler对象啦!日后msg.target.dispatchMessage()时会使用。  
  4.     msg.target = this;    
  5.   
  6.     if (mAsynchronous) {  
  7.         msg.setAsynchronous(true);  
  8.     }  
  9.     return queue.enqueueMessage(msg, uptimeMillis);  
  10. }  

请大家注意msg.target = this;一句,记录的就是handler对象。

当Looper的消息循环最终调用到msg.target.dispatchMessage()时,会间接调用到handler的handleMessage()函数,从而对消息进行实际处理。

在实际运用handler时,大体有两种方式。一种方式是写一个继承于Handler的新类,并在新类里实现自己的handleMessage()成员函数;另一种方式是在创建匿名Handler对象时,直接修改handleMessage()成员函数。

4消息队列MessageQueue

在刚刚介绍Handler的sendMessageAtTime()时,我们已经看到最终会调用queue.enqueueMessage()来向消息队列打入消息。queue对应的类是MessageQueue,其定义截选如下:
【frameworks/base/core/java/android/os/MessageQueue.java】

[java] view plain copy
 
  1. public final class MessageQueue {  
  2.     // True if the message queue can be quit.  
  3.     private final boolean mQuitAllowed;  
  4.   
  5.     @SuppressWarnings("unused")  
  6.     private int mPtr; // used by native code  
  7.   
  8.     Message mMessages;  // 消息队列!  
  9.     private final ArrayList<IdleHandler> mIdleHandlers = new ArrayList<IdleHandler>();  
  10.     private IdleHandler[] mPendingIdleHandlers;  
  11.     private boolean mQuitting;  
  12.   
  13.     // Indicates whether next() is blocked waiting in pollOnce() with a non-zero timeout.  
  14.     private boolean mBlocked;  
  15.   
  16.     // The next barrier token.  
  17.     // Barriers are indicated by messages with a null target whose arg1 field carries the token.  
  18.     private int mNextBarrierToken;  
  19.   
  20.     private native static int nativeInit();  
  21.     private native static void nativeDestroy(int ptr);  
  22.     private native static void nativePollOnce(int ptr, int timeoutMillis);  
  23.     private native static void nativeWake(int ptr);  
  24.     private native static boolean nativeIsIdling(int ptr);  
  25.     . . . . . .  

其中Message mMessages记录的就是一条消息链表。另外还有几个native函数,这就说明MessageQueue会通过JNI技术调用到底层代码。mMessages域记录着消息队列中所有Java层的实质消息。请大家注意,记录的只是Java层的消息,不包括C++层的。MessageQueue的示意图如下:

技术分享

 

 


4.1打入消息

4.1.1enqueueMessage()

很明显,enqueueMessage()就是在向MessageQueue的消息链表里插入Message。其代码截选如下:
【frameworks/base/core/java/android/os/MessageQueue.java】

[java] view plain copy
 
  1. boolean enqueueMessage(Message msg, long when) {  
  2.     . . . . . .  
  3.         . . . . . .  
  4.         msg.when = when;  
  5.         Message p = mMessages;  
  6.         boolean needWake;  
  7.         if (p == null || when == 0 || when < p.when) {  
  8.             // 此时,新消息会插入到链表的表头,这意味着队列需要调整唤醒时间啦。  
  9.             msg.next = p;  
  10.             mMessages = msg;  
  11.             needWake = mBlocked;  
  12.         } else {  
  13.             // 此时,新消息会插入到链表的内部,一般情况下,这不需要调整唤醒时间。  
  14.             // 但还必须考虑到当表头为“同步分割栏”的情况  
  15.             needWake = mBlocked && p.target == null && msg.isAsynchronous();  
  16.             Message prev;  
  17.             for (;;) {  
  18.                 prev = p;  
  19.                 p = p.next;  
  20.                 if (p == null || when < p.when) {  
  21.                     break;  
  22.                 }  
  23.                 if (needWake && p.isAsynchronous()) {  
  24.                     // 说明即便msg是异步的,也不是链表中第一个异步消息,所以没必要唤醒了  
  25.                     needWake = false;    
  26.                 }  
  27.             }  
  28.             msg.next = p;  
  29.             prev.next = msg;  
  30.         }  
  31.   
  32.         if (needWake) {  
  33.             nativeWake(mPtr);  
  34.         }  
  35.     . . . . . .  
  36. }  

打入消息的动作并不复杂,无非是在消息链表中找到合适的位置,插入Message节点而已。因为消息链表是按时间进行排序的,所以主要是在比对Message携带的when信息。消息链表的首个节点对应着最先将被处理的消息,如果Message被插到链表的头部了,就意味着队列的最近唤醒时间也应该被调整了,因此needWake会被设为true,以便代码下方可以走进nativeWake()。

4.1.2说说“同步分割栏”

上面的代码中还有一个“同步分割栏”的概念需要提一下。所谓“同步分割栏”,可以被理解为一个特殊Message,它的target域为null。它不能通过sendMessageAtTime()等函数打入到消息队列里,而只能通过调用Looper的postSyncBarrier()来打入。

“同步分割栏”是起什么作用的呢?它就像一个卡子,卡在消息链表中的某个位置,当消息循环不断从消息链表中摘取消息并进行处理时,一旦遇到这种“同步分割栏”,那么即使在分割栏之后还有若干已经到时的普通Message,也不会摘取这些消息了。请注意,此时只是不会摘取“普通Message”了,如果队列中还设置有“异步Message”,那么还是会摘取已到时的“异步Message”的。

在Android的消息机制里,“普通Message”和“异步Message”也就是这点儿区别啦,也就是说,如果消息列表中根本没有设置“同步分割栏”的话,那么“普通Message”和“异步Message”的处理就没什么大的不同了。

打入“同步分割栏”的postSyncBarrier()函数的代码如下:
【frameworks/base/core/java/android/os/Looper.java】

[java] view plain copy
 
  1. public int postSyncBarrier() {  
  2.     return mQueue.enqueueSyncBarrier(SystemClock.uptimeMillis());  
  3. }  

【frameworks/base/core/java/android/os/MessageQueue.java】

[java] view plain copy
 
  1. int enqueueSyncBarrier(long when) {  
  2.     synchronized (this) {  
  3.         final int token = mNextBarrierToken++;  
  4.         final Message msg = Message.obtain();  
  5.         msg.when = when;  
  6.         msg.arg1 = token;  
  7.   
  8.         Message prev = null;  
  9.         Message p = mMessages;  
  10.         if (when != 0) {  
  11.             while (p != null && p.when <= when) {  
  12.                 prev = p;  
  13.                 p = p.next;  
  14.             }  
  15.         }  
  16.         if (prev != null) {   
  17.             msg.next = p;  
  18.             prev.next = msg;  
  19.         } else {  
  20.             msg.next = p;  
  21.             mMessages = msg;  
  22.         }  
  23.         return token;  
  24.     }  
  25. }  

要得到“异步Message”,只需调用一下Message的setAsynchronous()即可:
【frameworks/base/core/java/android/os/Message.java】

[java] view plain copy
 
  1. public void setAsynchronous(boolean async) {  
  2.     if (async) {  
  3.         flags |= FLAG_ASYNCHRONOUS;  
  4.     } else {  
  5.         flags &= ~FLAG_ASYNCHRONOUS;  
  6.     }  
  7. }  

一般,我们是通过“异步Handler”向消息队列打入“异步Message”的。异步Handler的mAsynchronous域为true,因此它在调用enqueueMessage()时,可以走入:

[java] view plain copy
 
  1. if (mAsynchronous) {  
  2.     msg.setAsynchronous(true);  
  3. }  

现在我们画一张关于“同步分割栏”的示意图:

技术分享

 

 

 

图中的消息队列中有一个“同步分割栏”,因此它后面的“2”号Message即使到时了,也不会摘取下来。而“3”号Message因为是个异步Message,所以当它到时后,是可以进行处理的。

“同步分割栏”这种卡子会一直卡在消息队列中,除非我们调用removeSyncBarrier()删除这个卡子。
【frameworks/base/core/java/android/os/Looper.java】

[java] view plain copy
 
  1. public void removeSyncBarrier(int token) {  
  2.     mQueue.removeSyncBarrier(token);  
  3. }  

【frameworks/base/core/java/android/os/MessageQueue.java】

[java] view plain copy
 
  1. void removeSyncBarrier(int token) {  
  2.     // Remove a sync barrier token from the queue.  
  3.     // If the queue is no longer stalled by a barrier then wake it.  
  4.     synchronized (this) {  
  5.         Message prev = null;  
  6.         Message p = mMessages;  
  7.         while (p != null && (p.target != null || p.arg1 != token)) {  
  8.             prev = p;  
  9.             p = p.next;  
  10.         }  
  11.         if (p == null) {  
  12.             throw new IllegalStateException("The specified message queue synchronization "  
  13.                     + " barrier token has not been posted or has already been removed.");  
  14.         }  
  15.         final boolean needWake;  
  16.         if (prev != null) {  
  17.             prev.next = p.next;  
  18.             needWake = false;  
  19.         } else {  
  20.             mMessages = p.next;  
  21.             needWake = mMessages == null || mMessages.target != null;  
  22.         }  
  23.         p.recycle();  
  24.   
  25.         // If the loop is quitting then it is already awake.  
  26.         // We can assume mPtr != 0 when mQuitting is false.  
  27.         if (needWake && !mQuitting) {  
  28.             nativeWake(mPtr);  
  29.         }  
  30.     }  
  31. }  

和插入消息类似,如果删除动作改变了链表的头部,也意味着队列的最近唤醒时间应该被调整了,因此needWake会被设为true,以便代码下方可以走进nativeWake()。

4.1.3nativeWake()

nativeWake()对应的C++层函数如下:
【frameworks/base/core/jni/android_os_MessageQueue.cpp】

[cpp] view plain copy
 
  1. static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jint ptr) {  
  2.     NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);  
  3.     return nativeMessageQueue->wake();  
  4. }  

 

[cpp] view plain copy
 
  1. void NativeMessageQueue::wake() {  
  2.     mLooper->wake();  
  3. }  

 

【system/core/libutils/Looper.cpp】

[cpp] view plain copy
 
  1. void Looper::wake() {  
  2.     . . . . . .  
  3.     ssize_t nWrite;  
  4.     do {  
  5.         nWrite = write(mWakeWritePipeFd, "W", 1);  
  6.     } while (nWrite == -1 && errno == EINTR);  
  7.   
  8.     if (nWrite != 1) {  
  9.         if (errno != EAGAIN) {  
  10.             ALOGW("Could not write wake signal, errno=%d", errno);  
  11.         }  
  12.     }  
  13. }  

wake()动作主要是向一个管道的“写入端”写入了“W”。有关这个管道的细节,我们会在后文再细说,这里先放下。

4.2消息循环

接下来我们来看看消息循环。我们从Looper的Loop()函数开始讲起。下面是loop()函数的简略代码,我们只保留了其中最关键的部分:
【frameworks/base/core/java/android/os/Looper.java】

[java] view plain copy
 
  1. public static void loop()   
  2. {  
  3.     final Looper me = myLooper();  
  4.     . . . . . .  
  5.     final MessageQueue queue = me.mQueue;  
  6.   
  7.     Binder.clearCallingIdentity();  
  8.     final long ident = Binder.clearCallingIdentity();  
  9.   
  10.     for (;;) {  
  11.         Message msg = queue.next(); // might block  
  12.         . . . . . .  
  13.         msg.target.dispatchMessage(msg);  // 派发消息  
  14.         . . . . . .  
  15.         final long newIdent = Binder.clearCallingIdentity();  
  16.         . . . . . .  
  17.         msg.recycle();  
  18.     }  
  19. }  

无非是在一个for循环里不断摘取队列里的下一条消息,而后dispatchMessage()消息。呃,至少逻辑上就是这么简单,但如果我们希望再探索得更深一点的话,就得详细研究MessageQueue以及其next()函数了。

 

 对于Looper而言,它主要关心的是从消息队列里摘取消息,而后分派消息。然而对消息队列而言,在摘取消息时还要考虑更多技术细节。它关心的细节有:
1)如果消息队列里目前没有合适的消息可以摘取,那么不能让它所属的线程“傻转”,而应该使之阻塞;
2)队列里的消息应该按其“到时”的顺序进行排列,最先到时的消息会放在队头,也就是mMessages域所指向的消息,其后的消息依次排开;
3)阻塞的时间最好能精确一点儿,所以如果暂时没有合适的消息节点可摘时,要考虑链表首个消息节点将在什么时候到时,所以这个消息节点距离当前时刻的时间差,就是我们要阻塞的时长。
4)有时候外界希望队列能在即将进入阻塞状态之前做一些动作,这些动作可以称为idle动作,我们需要兼顾处理这些idle动作。一个典型的例子是外界希望队列在进入阻塞之前做一次垃圾收集。

以上所述的细节,基本上都体现在MessageQueue的next()函数里了,现在我们就来看这个函数的主要流程。

4.2.1MessageQueue的next()成员函数

MessageQueue的next()函数的代码截选如下:

[java] view plain copy
 
  1. Message next()   
  2. {  
  3.     int pendingIdleHandlerCount = -1; // -1 only during first iteration  
  4.     int nextPollTimeoutMillis = 0;  
  5.       
  6.     for (;;) {  
  7.         . . . . . .  
  8.         nativePollOnce(mPtr, nextPollTimeoutMillis);    // 阻塞于此  
  9.         . . . . . .  
  10.             // 获取next消息,如能得到就返回之。  
  11.             final long now = SystemClock.uptimeMillis();  
  12.             Message prevMsg = null;  
  13.             Message msg = mMessages;  // 先尝试拿消息队列里当前第一个消息  
  14.               
  15.             if (msg != null && msg.target == null) {  
  16.                 // 如果从队列里拿到的msg是个“同步分割栏”,那么就寻找其后第一个“异步消息”  
  17.                 do {  
  18.                     prevMsg = msg;  
  19.                     msg = msg.next;  
  20.                 } while (msg != null && !msg.isAsynchronous());  
  21.             }  
  22.               
  23.             if (msg != null) {  
  24.                 if (now < msg.when) {  
  25.                     // Next message is not ready.  Set a timeout to wake up when it is ready.  
  26.                     nextPollTimeoutMillis = (int) Math.min(msg.when - now,   
  27.                                                                    Integer.MAX_VALUE);  
  28.                 } else {  
  29.                     // Got a message.  
  30.                     mBlocked = false;  
  31.                     if (prevMsg != null) {  
  32.                         prevMsg.next = msg.next;  
  33.                     } else {  
  34.                         mMessages = msg.next;  // 重新设置一下消息队列的头部  
  35.                     }  
  36.                     msg.next = null;  
  37.                     if (false) Log.v("MessageQueue", "Returning message: " + msg);  
  38.                     msg.markInUse();  
  39.                     return msg;     // 返回得到的消息对象  
  40.                 }  
  41.             } else {  
  42.                 // No more messages.  
  43.                 nextPollTimeoutMillis = -1;  
  44.             }  
  45.   
  46.             // Process the quit message now that all pending messages have been handled.  
  47.             if (mQuitting) {  
  48.                 dispose();  
  49.                 return null;  
  50.             }  
  51.             if (pendingIdleHandlerCount < 0  
  52.                         && (mMessages == null || now < mMessages.when)) {  
  53.                     pendingIdleHandlerCount = mIdleHandlers.size();  
  54.             }  
  55.             if (pendingIdleHandlerCount <= 0) {  
  56.                 // No idle handlers to run.  Loop and wait some more.  
  57.                 mBlocked = true;  
  58.                 continue;  
  59.             }  
  60.         . . . . . .  
  61.         // 处理idle handlers部分  
  62.         for (int i = 0; i < pendingIdleHandlerCount; i++) {  
  63.             final IdleHandler idler = mPendingIdleHandlers[i];  
  64.             mPendingIdleHandlers[i] = null; // release the reference to the handler  
  65.   
  66.             boolean keep = false;  
  67.             try {  
  68.                 keep = idler.queueIdle();  
  69.             } catch (Throwable t) {  
  70.                 Log.wtf("MessageQueue", "IdleHandler threw exception", t);  
  71.             }  
  72.   
  73.             if (!keep) {  
  74.                 synchronized (this) {  
  75.                     mIdleHandlers.remove(idler);  
  76.                 }  
  77.             }  
  78.         }  
  79.           
  80.         pendingIdleHandlerCount = 0;  
  81.         nextPollTimeoutMillis = 0;  
  82.     }  
  83. }  

这个函数里的for循环并不是起循环摘取消息节点的作用,而是为了连贯“当前时间点”和“处理下一条消息的时间点”。简单地说,当“定时机制”触发“摘取一条消息”的动作时,会判断事件队列的首条消息是否真的到时了,如果已经到时了,就直接返回这个msg,而如果尚未到时,则会努力计算一个较精确的等待时间(nextPollTimeoutMillis),计算完后,那个for循环会掉过头再次调用到nativePollOnce(mPtr, nextPollTimeoutMillis),进入阻塞状态,从而等待合适的时长。

上面代码中也处理了“同步分割栏”的情况。如果从队列里获取的消息是个“同步分割栏”的话,可千万不能把“同步分割栏”给返回了,此时会尝试找寻其后第一个“异步消息”。

next()里另一个要说的是那些Idle Handler,当消息队列中没有消息需要马上处理时,会判断用户是否设置了Idle Handler,如果有的话,则会尝试处理mIdleHandlers中所记录的所有Idle Handler,此时会逐个调用这些Idle Handler的queueIdle()成员函数。我们举一个例子,在ActivityThread中,在某种情况下会在消息队列中设置GcIdler,进行垃圾收集,其定义如下:

[java] view plain copy
 
  1. final class GcIdler implements MessageQueue.IdleHandler {  
  2.     @Override  
  3.     public final boolean queueIdle() {  
  4.         doGcIfNeeded();  
  5.         return false;  
  6.     }  
  7. }  

一旦队列里设置了这个Idle Handler,那么当队列中没有马上需处理的消息时,就会进行垃圾收集。

 

4.2.1.1nativePollOnce()

前文我们已经说过,next()中调用的nativePollOnce()起到了阻塞作用,保证消息循环不会在无消息处理时一直在那里“傻转”。那么,nativePollOnce()函数究竟是如何实现阻塞功能的呢?我们来探索一下。首先,MessageQueue类里声明的几个native函数,对应的JNI实现位于android_os_MessageQueue.cpp文件中:
【frameworks/base/core/jni/android_os_MessageQueue.cpp】

[cpp] view plain copy
 
  1. static JNINativeMethod gMessageQueueMethods[] = {  
  2.     /* name, signature, funcPtr */  
  3.     { "nativeInit", "()I", (void*)android_os_MessageQueue_nativeInit },  
  4.     { "nativeDestroy", "(I)V", (void*)android_os_MessageQueue_nativeDestroy },  
  5.     { "nativePollOnce", "(II)V", (void*)android_os_MessageQueue_nativePollOnce },  
  6.     { "nativeWake", "(I)V", (void*)android_os_MessageQueue_nativeWake },  
  7.     { "nativeIsIdling", "(I)Z", (void*)android_os_MessageQueue_nativeIsIdling }  
  8. };  

而且在MessageQueue构造之时,就会调用nativeInit()函数。

 

目前我们只关心nativePollOnce对应的android_os_MessageQueue_nativePollOnce()。其代码如下:

[cpp] view plain copy
 
  1. static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jclass clazz,  
  2.                                                              jint ptr, jint timeoutMillis)   
  3. {  
  4.     NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);  
  5.     nativeMessageQueue->pollOnce(env, timeoutMillis);  
  6. }  

看到了吧,ptr参数会被强制转换成NativeMessageQueue*。

 

NativeMessageQueue的pollOnce()如下:
【frameworks/base/core/jni/android_os_MessageQueue.cpp】

[cpp] view plain copy
 
  1. void NativeMessageQueue::pollOnce(JNIEnv* env, int timeoutMillis) {  
  2.     mInCallback = true;  
  3.     mLooper->pollOnce(timeoutMillis);   // 用到C++层的Looper对象  
  4.     mInCallback = false;  
  5.     if (mExceptionObj) {  
  6.         env->Throw(mExceptionObj);  
  7.         env->DeleteLocalRef(mExceptionObj);  
  8.         mExceptionObj = NULL;  
  9.     }  
  10. }  

这里会用到C++层的Looper类,它和Java层的Looper类可是不一样的哩。C++层的Looper类的定义截选如下:
【system/core/include/utils/Looper.h】

[cpp] view plain copy
 
  1. class Looper : public ALooper, public RefBase {  
  2. protected:  
  3.     virtual ~Looper();  
  4.   
  5. public:  
  6.     Looper(bool allowNonCallbacks);  
  7.     bool getAllowNonCallbacks() const;  
  8.     int pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData);  
  9.     . . . . . .  
  10.     int pollAll(int timeoutMillis, int* outFd, int* outEvents, void** outData);  
  11.     . . . . . .  
  12.     void wake();  
  13.   
  14.     int addFd(int fd, int ident, int events, ALooper_callbackFunc callback, void* data);  
  15.     int addFd(int fd, int ident, int events, const sp<LooperCallback>& callback, void* data);  
  16.     int removeFd(int fd);  
  17.   
  18.     void sendMessage(const sp<MessageHandler>& handler, const Message& message);  
  19.     void sendMessageDelayed(nsecs_t uptimeDelay, const sp<MessageHandler>& handler,  
  20.             const Message& message);  
  21.     void sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler>& handler,  
  22.             const Message& message);  
  23.     void removeMessages(const sp<MessageHandler>& handler);  
  24.     void removeMessages(const sp<MessageHandler>& handler, int what);  
  25.   
  26.     bool isIdling() const;  
  27.     static sp<Looper> prepare(int opts);  
  28.     static void setForThread(const sp<Looper>& looper);  
  29.     static sp<Looper> getForThread();  
  30.     . . . . . .  
  31.     . . . . . .  
  32. };  

我们把C++层的NativeMessageQueue和Looper融入前文的示意图,可以得到一张新的示意图,如下所示:

技术分享

 

 

C++层的Looper的构造函数如下:

[cpp] view plain copy
 
  1. Looper::Looper(bool allowNonCallbacks) :  
  2.         mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),  
  3.         mResponseIndex(0), mNextMessageUptime(LLONG_MAX)   
  4. {  
  5.     int wakeFds[2];  
  6.     int result = pipe(wakeFds);  // 创建一个管道  
  7.     LOG_ALWAYS_FATAL_IF(result != 0, "Could not create wake pipe.  errno=%d", errno);  
  8.   
  9.     mWakeReadPipeFd = wakeFds[0];    // 管道的“读取端”  
  10.     mWakeWritePipeFd = wakeFds[1];   // 管道的“写入端”  
  11.   
  12.     result = fcntl(mWakeReadPipeFd, F_SETFL, O_NONBLOCK);  
  13.     LOG_ALWAYS_FATAL_IF(result != 0,   
  14.                        "Could not make wake read pipe non-blocking.  errno=%d", errno);  
  15.   
  16.     result = fcntl(mWakeWritePipeFd, F_SETFL, O_NONBLOCK);  
  17.     LOG_ALWAYS_FATAL_IF(result != 0,   
  18.                        "Could not make wake write pipe non-blocking.  errno=%d", errno);  
  19.     mIdling = false;  
  20.   
  21.     // 创建一个epoll  
  22.     mEpollFd = epoll_create(EPOLL_SIZE_HINT);  
  23.     LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance.  errno=%d", errno);  
  24.   
  25.     struct epoll_event eventItem;  
  26.     memset(& eventItem, 0, sizeof(epoll_event));   
  27.     eventItem.events = EPOLLIN;  
  28.     eventItem.data.fd = mWakeReadPipeFd;     
  29.     // 监听管道的read端  
  30.     result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeReadPipeFd, & eventItem);  
  31.     LOG_ALWAYS_FATAL_IF(result != 0,   
  32.                      "Could not add wake read pipe to epoll instance.  errno=%d", errno);  
  33. }  

可以看到在构造Looper对象时,其内部除了创建了一个管道以外,还创建了一个epoll来监听管道的“读取端”。也就是说,是利用epoll机制来完成阻塞动作的。每当我们向消息队列发送事件时,最终会间接向管道的“写入端”写入数据,这个前文已有叙述,于是epoll通过管道的“读取端”立即就感知到了风吹草动,epoll_wait()在等到事件后,随即进行相应的事件处理。这就是消息循环阻塞并处理的大体流程。当然,因为向管道写数据只是为了通知风吹草动,所以写入的数据是非常简单的“W”字符串。现在大家不妨再看看前文阐述“nativeWake()”的小节,应该能明白了吧。

 

我们还是继续说消息循环。Looper的pollOnce()函数如下:
【system/core/libutils/Looper.cpp】

[cpp] view plain copy
 
  1. int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData)   
  2. {  
  3.     int result = 0;  
  4.     for (;;) {  
  5.         . . . . . .  
  6.         if (result != 0) {  
  7.             . . . . . .  
  8.             if (outFd != NULL) *outFd = 0;  
  9.             if (outEvents != NULL) *outEvents = 0;  
  10.             if (outData != NULL) *outData = NULL;  
  11.             return result;  
  12.         }  
  13.   
  14.         result = pollInner(timeoutMillis);  
  15.     }  
  16. }  

 

[cpp] view plain copy
 
  1. int Looper::pollInner(int timeoutMillis)   
  2. {  
  3. . . . . . .  
  4.     // 阻塞、等待  
  5. int eventCount = epoll_wait( mEpollFd, eventItems,   
  6.                                    EPOLL_MAX_EVENTS, timeoutMillis);  
  7.     . . . . . .  
  8.     . . . . . .  
  9.     // 处理所有epoll事件  
  10.     for (int i = 0; i < eventCount; i++)   
  11.     {  
  12.         int fd = eventItems[i].data.fd;  
  13.         uint32_t epollEvents = eventItems[i].events;  
  14.         if (fd == mWakeReadPipeFd)   
  15.         {  
  16.             if (epollEvents & EPOLLIN) {  
  17.                 awoken();  // 从管道中感知到EPOLLIN,于是调用awoken()  
  18.             }   
  19.             . . . . . .  
  20.         }   
  21.         else   
  22.         {  
  23.             // 如果是除管道以外的其他fd发生了变动,那么根据其对应的request,  
  24.             // 将response先记录进mResponses  
  25.             ssize_t requestIndex = mRequests.indexOfKey(fd);  
  26.             if (requestIndex >= 0) {  
  27.                 int events = 0;  
  28.                 if (epollEvents & EPOLLIN ) events |= ALOOPER_EVENT_INPUT;  
  29.                 if (epollEvents & EPOLLOUT) events |= ALOOPER_EVENT_OUTPUT;  
  30.                 if (epollEvents & EPOLLERR) events |= ALOOPER_EVENT_ERROR;  
  31.                 if (epollEvents & EPOLLHUP) events |= ALOOPER_EVENT_HANGUP;  
  32.                 // 内部会调用 mResponses.push(response);  
  33.                 pushResponse(events, mRequests.valueAt(requestIndex));  
  34.             }   
  35.             . . . . . .  
  36.         }  
  37.     }  
  38.       
  39. Done: ;  
  40.     . . . . . .  
  41.     // 调用尚未处理的事件的回调  
  42.     while (mMessageEnvelopes.size() != 0)   
  43.     {  
  44.         nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);  
  45.         const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0);  
  46.         if (messageEnvelope.uptime <= now)   
  47.         {  
  48.             {   
  49.                 sp<MessageHandler> handler = messageEnvelope.handler;  
  50.                 Message message = messageEnvelope.message;  
  51.                 mMessageEnvelopes.removeAt(0);  
  52.                 . . . . . .  
  53.                 handler->handleMessage(message);  
  54.             }  
  55.             . . . . . .  
  56.         }   
  57.         else {  
  58.             mNextMessageUptime = messageEnvelope.uptime;  
  59.             break;  
  60.         }  
  61.     }  
  62.       
  63.     . . . . . .  
  64.     // 调用所有response记录的回调  
  65.     for (size_t i = 0; i < mResponses.size(); i++) {  
  66.         Response& response = mResponses.editItemAt(i);  
  67.         if (response.request.ident == ALOOPER_POLL_CALLBACK) {  
  68.             . . . . . .  
  69.             int callbackResult = response.request.callback->handleEvent(fd, events, data);  
  70.             if (callbackResult == 0) {  
  71.                 removeFd(fd);  
  72.             }  
  73.             . . . . . .  
  74.         }  
  75.     }  
  76.     return result;  
  77. }  

现在我们可以画一张调用示意图,理一下loop()函数的调用关系,如下:

技术分享

 

 

pollInner()调用epoll_wait()时传入的timeoutMillis参数,其实来自于前文所说的MessageQueue的next()函数里的nextPollTimeoutMillis,next()函数里在以下3种情况下,会给nextPollTimeoutMillis赋不同的值:
1)如果消息队列中的下一条消息还要等一段时间才到时的话,那么nextPollTimeoutMillis赋值为Math.min(msg.when - now, Integer.MAX_VALUE),即时间差;
2)如果消息队列已经是空队列了,那么nextPollTimeoutMillis赋值为-1;
3)不管前两种情况下是否已给nextPollTimeoutMillis赋过值了,只要队列中有Idle Handler需要处理,那么在处理完所有Idle Handler之后,会强制将nextPollTimeoutMillis赋值为0。这主要是考虑到在处理Idle Handler时,不知道会耗时多少,而在此期间消息队列的“到时情况”有可能已发生改变。

不管epoll_wait()的超时阀值被设置成什么,只要程序从epoll_wait()中返回,就会尝试处理等到的epoll事件。目前我们的主要关心点是事件机制,所以主要讨论当fd 等于mWakeReadPipeFd时的情况,此时会调用一下awoken()函数。该函数很简单,只是在读取mWakeReadPipeFd而已:

[cpp] view plain copy
 
  1. void Looper::awoken() {  
  2. #if DEBUG_POLL_AND_WAKE  
  3.     ALOGD("%p ~ awoken", this);  
  4. #endif  
  5.   
  6.     char buffer[16];  
  7.     ssize_t nRead;  
  8.     do {  
  9.         nRead = read(mWakeReadPipeFd, buffer, sizeof(buffer));  
  10.     } while ((nRead == -1 && errno == EINTR) || nRead == sizeof(buffer));  
  11. }  

为什么要起个名字叫awoken()呢?这是因为当初发送事件时,最终是调用一个wake()函数来通知消息队列的,现在epoll_wait()既然已经感应到了,自然相当于“被唤醒”(awoken)了。

 

除了感知mWakeReadPipeFd管道的情况以外,epoll还会感知其他一些fd对应的事件。在Looper中有一个mRequests键值向量表(KeyedVector<int, Request> mRequests),其键值就是感兴趣的fd。如果收到的epoll事件所携带的fd可以在这张表里查到,那么就将该fd对应的Request整理进Response对象,并将该Response对象记入mResponses表。在pollInner()的最后,会用一个for循环遍历mResponses表,分析每个Response表项对应的Request是不是需要callback,如果需要的话,执行对应的回调函数:

[cpp] view plain copy
 
  1. int callbackResult = response.request.callback->handleEvent(fd, events, data);  
  2. if (callbackResult == 0) {  
  3.     removeFd(fd);  
  4. }  

可以看到,handleEvent()的返回值将决定那个Request表项是否继续保留在mRequests表中,如果返回值为0,说明不必保留了,所以删除之。删除时会同时从epoll中注销这个Request对应的fd,表示不再对这个fd感兴趣了。

 

pollInner()内部还会集中处理所记录的所有C++层的Message。在一个while循环中,不断摘取mMessageEnvelopes向量表的第0个MessageEnvelope,如果消息已经到时,则回调handleMessage()。

[cpp] view plain copy
 
  1. sp<MessageHandler> handler = messageEnvelope.handler;  
  2. Message message = messageEnvelope.message;  
  3. mMessageEnvelopes.removeAt(0);  
  4. . . . . . .  
  5. handler->handleMessage(message);  

而如果消息未到时,说明while循环可以break了。

 

C++层的Looper及这个层次的消息链表,再加上对应其他fd的Request和Response,可以形成下面这张示意图:

技术分享

 

 

 从我们的分析中可以知道,在Android中,不光是Java层可以发送Message,C++层也可以发送,当然,不同层次的Message是放在不同层次的消息链中的。在Java层,每次尝试从队列中获取一个Message,而后dispatch它。而C++层的消息则尽量在一次pollOnce中集中处理完毕,这是它们的一点不同。

5尾声

关于Android的消息机制,我们就先说这么多。总体上的而言还是比较简单的,无非是通过Handler向Looper的消息队列中插入Message,而后再由Looper在消息循环里具体处理。因为消息队列本身不具有链表一变动就能马上感知的功能,所以它需要借助管道和epoll机制来监听变动。当外界向消息队列中打入新消息后,就向管道的“写入端”写入简单数据,于是epoll可以立即感知到管道的变动,从何激发从消息队列中摘取消息的动作。这就是Android消息机制的大体情况。

聊一聊Android的消息机制