[Android源代码分析]Android消息机制，Handler，Message，Looper，MessageQueue
Posted OSTCB
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了[Android源代码分析]Android消息机制，Handler，Message，Looper，MessageQueue相关的知识，希望对你有一定的参考价值。
最近准备把android源码大致过一遍，不敢私藏，写出来分享给大家，顺便记录一下自己的学习感悟。里面一定有一些错误的地方，希望广大看客理解理解。
网上也有不少分析文章，这里我尽量分析的更加细致详尽。不留死角。
一.核心循环体：Looper.loop();
我们知道，在线程run()中Looper.prepare();Looper.looper()。之后这个线程就是一个HandlerThread了。我们可以通过Handler在另外一个线程中(自己也可以)向这个线程发送消息，在这个线程中处理消息。
简单来说，Looper.loop(),就是一个循环体，循环着从MessageQueue中取消息，处理消息。现在存在几个问题。这就是整个消息处理框架的核心所在。框架所有的Api(消息的增删改查)都是为了这个循环体服务的。当然简单的for(;;)是远远不够的，框架需要考虑到以下问题：
1. 线程安全问题：消息的增删改查是多线程环境，所以要保证整个消息队列的线程安全。
2. 线程阻塞问题：当队列中没有到期需要处理的消息时，怎样避免线程空转浪费性能。
3. 与native层交互问题：我们知道，Android框架层分java和native层，基本我们平时常用在java层常用的核心框架，在native层都有另外一套与之匹配的“分身”，而且java层的众多功能实现都要依赖native层。
4. 性能问题：比如说Message对象的生灭在这个框架中非常频繁，可以使用对象缓冲池来提高性能，减轻gc压力。
1.Looper.looper()的源码(Android 6.0)
Looper.java(frameworks/base/core/java/android/os/Looper.java)
/**
* Run the message queue in this thread. Be sure to call
* {@link #quit()} to end the loop.
*/
public static void loop() {
//1.指向static final ThreadLocal<Looper> sThreadLocal.get()
//相当于一个Map key是Thread.currentThread()
//ThreadLocal知识:http://www.iteye.com/topic/103804
//只是一个本线程持有的变量表而已，随线程生灭,
//这里只是取出当前线程对应的looper对象
final Looper me = myLooper();
//2.如果这个线程没有对应的looper对象，即没有Looper.prepare();则抛出异常
if (me == null) {
throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");
}
//3.从Looper中取出MessageQueue，一个Looper持有一个MessageQueue
final MessageQueue queue = me.mQueue;
// Make sure the identity of this thread is that of the local process,
// and keep track of what that identity token actually is.
//确保这个线程是本地的，因为Handler Message可用于IPC(猜测)
Binder.clearCallingIdentity();
final long ident = Binder.clearCallingIdentity();
//4.循环体
for (;;) {
//5.循环取出下一条消息来处理，会在这里阻塞住，这里可以猜到消息队列时一个链表结构，先剧透一下上面说的线程阻塞，防止空转也是在这个函数里实现的。下面会具体分析。
Message msg = queue.next(); // might block
//没有message即返回
if (msg == null) {
// No message indicates that the message queue is quitting.
return;
}
// This must be in a local variable, in case a UI event sets the logger
Printer logging = me.mLogging;
if (logging != null) {
logging.println(">>>>> Dispatching to " + msg.target + " " +
msg.callback + ": " + msg.what);
}
//6.消息分发处理，处理消息，最终调用用户实现的handlerMessage()的地方。下面分析
msg.target.dispatchMessage(msg);
if (logging != null) {
logging.println("<<<<< Finished to " + msg.target + " " + msg.callback);
}
//7.翻译过来就是：确保在调度过程中的线程的身份没有被损坏，下面再细说
// Make sure that during the course of dispatching the
// identity of the thread wasn't corrupted.
final long newIdent = Binder.clearCallingIdentity();
if (ident != newIdent) {
Log.wtf(TAG, "Thread identity changed from 0x"
+ Long.toHexString(ident) + " to 0x"
+ Long.toHexString(newIdent) + " while dispatching to "
+ msg.target.getClass().getName() + " "
+ msg.callback + " what=" + msg.what);
}
//将处理完的Message对象回收，以备下次obtain重复使用。
msg.recycleUnchecked();
}
}Looper.prepare()最终调用
private static void prepare(boolean quitAllowed) {
if (sThreadLocal.get() != null) {
throw new RuntimeException("Only one Looper may be created per thread");
}
//为当前线程创建Looper对象，塞进ThreadLocal中
sThreadLocal.set(new Looper(quitAllowed));
}此外列出Looper类的成员变量
private static final String TAG = "Looper";
// sThreadLocal.get() will return null unless you've called prepare().
//每个线程对应一个Looper，对应多个handler。当在某个线程中调用Looper.Mylooper()时，
//调用sThreadLocal.get()返回对应该线程的Looper。
static final ThreadLocal<Looper> sThreadLocal = new ThreadLocal<Looper>();
//全局唯一的主线程对应的MainLooper
private static Looper sMainLooper; // guarded by Looper.class
//每个Looper管理一个MessageQueue
final MessageQueue mQueue;
final Thread mThread;
private Printer mLogging;Looper.java基本结束。
下面通过Message的增删查处理的顺序分析整个框架。
1）.下面先说Message的增
api:增的api对应有Handler.sendMessageXXXX,postXXXX,等。
Handler.java(frameworks/base/core/java/android/os/Handler.java)
a.首先得说Handler的初始化：
Handler3个构造函数最终调用两个版本。
public Handler(Callback callback, boolean async) {
//如果为真，则检查内存泄漏，就是判断自身是否为static类型
if (FIND_POTENTIAL_LEAKS) {
final Class<? extends Handler> klass = getClass();
if ((klass.isAnonymousClass() || klass.isMemberClass() || klass.isLocalClass()) &&
(klass.getModifiers() & Modifier.STATIC) == 0) {
Log.w(TAG, "The following Handler class should be static or leaks might occur: " +
klass.getCanonicalName());
}
}
mLooper = Looper.myLooper();
if (mLooper == null) {
throw new RuntimeException(
"Can't create handler inside thread that has not called Looper.prepare()");
}
mQueue = mLooper.mQueue;
mCallback = callback;
mAsynchronous = async;
}注意这个参数
/*
* Set this flag to true to detect anonymous, local or member classes
* that extend this Handler class and that are not static. These kind
* of classes can potentially create leaks.
*/
private static final boolean FIND_POTENTIAL_LEAKS = false;
这个参数用来检查可能的内存泄漏，设置这个为true的话，在构造方法中会检查，自身是否是静态的内部类，klass.isAnonymousClass() || klass.isMemberClass() || klass.isLocalClass()) &&(klass.getModifiers() & Modifier.STATIC，如果不是就抛出异常。关于内存泄漏，参看我另一篇文章链接。
剩下的就是简单的赋值了。
public Handler(Looper looper, Callback callback, boolean async) {
mLooper = looper;
mQueue = looper.mQueue;
mCallback = callback;
mAsynchronous = async;
}private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {
msg.target = this;
if (mAsynchronous) {
msg.setAsynchronous(true);
}
return queue.enqueueMessage(msg, uptimeMillis);
}还要注意的是async参数，如果你传入true，在最后向MessageQueue塞Message的时候会把这个Message设置为异步消息。
什么是异步消息？先剧透一下，后面要说到一个“同步屏障”的概念，Message链表一旦被插入一个“同步屏障”，那么屏障之后的所有同步消息将不会被处理，哪怕已经到期了。而异步消息则不受影响。
插入动作最后会通过setAsynchronous调用到MessageQueue方法的enqueueSyncBarreir(long when)方法，我们在下面分析。
b.现在拿到Handler对象了，看一下Handler这几个核心成员变量，可以看到Handler持有一个MessageQueue的引用，一个Looper的引用，这里的callback，有点像Thread和Runnable的关系，常见的设计模式，指向了用户handlerMessage的具体实现。
final MessageQueue mQueue;
final Looper mLooper;
//对应handlerMessage的具体实现
final Callback mCallback;
final boolean mAsynchronous;
//IPC相关
IMessenger mMessenger;说一说IMessenger，IPC相关，相信大家也看的出来，Handler也关联了Binder提供的IPC服务，Messenger关联一个可以发送消息的Handler。通过指定一Handler对象创建一个Messenger可以实现基于消息的夸进程通信。
线程和进程主要的差别无非是进程有独立的内存空间，一旦一个进程将Message通过Binder发送给另外一个进程中的线程的MessageQueue中，那么，这个Message一样可以和普通的Message统一处理了。可以说没毛病。
这方面在后面会具体讨论，现在先简单说一下。
c.handler有了，我们需要获得一个Message对象，上面说过了，对于Message这种大量朝生夕灭的对象，需要使用对象缓冲池，或者原型模式这些来优化性能，gc最烦的就是这种大量的“生的快，死的早”的对象。我们来看一下Message具体是怎么做的。
先看一下Message的成员变量，基本上都是熟悉的面孔
Message.java(frameworks/base/core/java/android/os/Message.java)
public int what;
public int arg1;
public int arg2;
public Object obj;
public Messenger replyTo;
public int sendingUid = -1;
*/
/*package*/ static final int FLAG_IN_USE = 1 <<0;
/*package*/ static final int FLAG_ASYNCHRONOUS = 1 <<1;
/** Flags to clear in the copyFrom method */
/*package*/ static final int FLAGS_TO_CLEAR_ON_COPY_FROM = FLAG_IN_USE;
/*package*/ int flags;
/*package*/ long when;
/*package*/ Bundle data;
/*package*/ Handler target;
/*package*/ Runnable callback;
// sometimes we store linked lists of these things
/*package*/ Message next;
private static final Object sPoolSync = new Object();
private static Message sPool;
private static int sPoolSize = 0;
private static final int MAX_POOL_SIZE = 50;
private static boolean gCheckRecycle = true;rePlyto是Messenger对象，和上面提到的IPC相关，是IPC的信使。
When这个比较重要，他代表了这个Message在什么时间需要被处理。
sPool就是Message对象缓冲池的链表头。
sPoolSync是同步块的标志，空对象。
MAX_POOL_SIZE，缓冲池对象最大数量
gCheckRecycle，好像没什么用，api大于等于21的时候，非法recycle Message对象的时候会多抛一个异常。
Message对象的获取，同步方法，需要保证Message链表的线程安全。
public static Message obtain() {
synchronized (sPoolSync) {
if (sPool != null) {
Message m = sPool;
sPool = m.next;
m.next = null;
m.flags = 0; // clear in-use flag
sPoolSize--;
return m;
}
}
//缓冲池用完即new新对象
return new Message();
}回收方法，较简单就不细说了，这里注意isInUse标志，在Message.next() 中会被置为true，表示Message正在使用，不能回收。
public void recycle() {
if (isInUse()) {
if (gCheckRecycle) {
throw new IllegalStateException("This message cannot be recycled because it "
+ "is still in use.");
}
return;
}
recycleUnchecked();
}
/**
* Recycles a Message that may be in-use.
* Used internally by the MessageQueue and Looper when disposing of queued Messages.
*/
//回收Message并清除里面的旧数据以备用。
void recycleUnchecked() {
// Mark the message as in use while it remains in the recycled object pool.
// Clear out all other details.
flags = FLAG_IN_USE;
what = 0;
arg1 = 0;
arg2 = 0;
obj = null;
replyTo = null;
sendingUid = -1;
when = 0;
target = null;
callback = null;
data = null;
synchronized (sPoolSync) {
if (sPoolSize <MAX_POOL_SIZE) {
next = sPool;
sPool = this;
sPoolSize++;
}
}
}d.现在Message对象拿到手了，下面分析Handler.sendMessage(msg)
Handler.java
无论是post(),sendXXX还是等等。都会封装到Message，计算时间后最终调用到sendMessageAtTime
public boolean sendMessageAtTime(Message msg, long uptimeMillis) {
MessageQueue queue = mQueue;
if (queue == null) {
RuntimeException e = new RuntimeException(
this + " sendMessageAtTime() called with no mQueue");
Log.w("Looper", e.getMessage(), e);
return false;
}
return enqueueMessage(queue, msg, uptimeMillis);
}最终调用MessageQueue的enqueueMessage方法
when在前面已经计算出来了，非常好算，加一下就好了
MessageQueue.java(frameworks/base/core/java/android/os/Message.java)
boolean enqueueMessage(Message msg, long when) {
if (msg.target == null) {
throw new IllegalArgumentException("Message must have a target.");
}
//确保msg不是正在使用的
if (msg.isInUse()) {
throw new IllegalStateException(msg + " This message is already in use.");
}
//同步方法，因为MessageQueue和Handler是1对N的关系，也就是说，会有多个线程可能同时调用该方法，所以需要同步处理一下。
synchronized (this) {
//mQuitting是队列退出的标志，调用quit方法退出队列将此标志置为true，接着looper就会退出，HandlerThread结束。
if (mQuitting) {
IllegalStateException e = new IllegalStateException(
msg.target + " sending message to a Handler on a dead thread");
Log.w(TAG, e.getMessage(), e);
msg.recycle();
return false;
}
//标记msg对象开始使用
msg.markInUse();
msg.when = when;
Message p = mMessages;
boolean needWake;
//当when=0／null或者when早于最前面一个msg的when的时候，将此msg插到队列最前面。这里的消息队列是按照when顺序排列的。如果这时队列是阻塞的话，即唤醒队列。
if (p == null || when == 0 || when < p.when) {
// New head, wake up the event queue if blocked.
msg.next = p;
mMessages = msg;
needWake = mBlocked;
} else {
// Inserted within the middle of the queue. Usually we don't have to wake
// up the event queue unless there is a barrier at the head of the queue
// and the message is the earliest asynchronous message in the queue.
//只有现在队列阻塞中并且,message对应的handler为空，并且Message为前面说的异步消息的时候，需要唤醒队列
needWake = mBlocked && p.target == null && msg.isAsynchronous();
//遍历队列根据when大小找到合适的地点插入。
Message prev;
for (;;) {
prev = p;
p = p.next;
if (p == null || when < p.when) {
break;
}
if (needWake && p.isAsynchronous()) {
needWake = false;
}
}
msg.next = p; // invariant: p == prev.next
prev.next = msg;
}
//根据结果判断是否唤醒队列(线程)，nativeWake是一个native方法，具体后面分析，mPtr是指向native层对应的MessageQueue对象的指针。
// We can assume mPtr != 0 because mQuitting is false.
if (needWake) {
nativeWake(mPtr);
}
}
return true;
}2.下面说说Message的查，即MessageQueue.next()方法，队列在这里阻塞，非常重要。
MessageQueue.java(frameworks/base/core/java/android/os/Message.java)
Message next() {
// Return here if the message loop has already quit and been disposed.
// This can happen if the application tries to restart a looper after quit
// which is not supported.
//娶到native层的MessageQueue对象指针
final long ptr = mPtr;
if (ptr == 0) {
return null;
}
int pendingIdleHandlerCount = -1; // -1 only during first iteration
int nextPollTimeoutMillis = 0;
for (;;) {
if (nextPollTimeoutMillis != 0) {
//去除所有在当前线程在内核中挂起的Binder命令，这个调用可能是有用的在做一个可能会阻塞线程很长一段时间的操作情况下，这是为了确保任何挂起的对象引用已被释放，以防止进程为保持对象的时间比它需要得长。
//简单的来说，下面当前线程就要调用nativePollOnce进入长时间的阻塞状态了，必须通知内核释放掉该进程在Binder驱动中挂起的对象，不然Binder将长时间无法获得来自该线程的通知对这些对象进行操作，造成Binder驱动的拥塞。
Binder.flushPendingCommands();
}
//阻塞函数，MessageQueue阻塞于此，实现延时阻塞的真正函数，这是个native函数，下面具体分析
nativePollOnce(ptr, nextPollTimeoutMillis);
//同步块，和上面enqueueMessage里的那个对应，保证Message链表结构的线程安全。
synchronized (this) {
// Try to retrieve the next message. Return if found.
final long now = SystemClock.uptimeMillis();
Message prevMsg = null;
Message msg = mMessages;
if (msg != null && msg.target == null) {
// Stalled by a barrier. Find the next asynchronous message in the queue.
//遍历Message链表找到一个非空并且不是异步的Message
do {
prevMsg = msg;
msg = msg.next;
} while (msg != null && !msg.isAsynchronous());
}
if (msg != null) {
//如果现在的时间早于Message需要被处理的时间
if (now < msg.when) {
// Next message is not ready. Set a timeout to wake up when it is ready.
//确定需要延时的时长，即msg的when-现在的时间，和Integer最大值的最大值
nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);
} else {
//不需要延时则可以直接获得Message,处理一下链表，标记Message的使用状态，返回Message给loop()去处理。
// Got a message.
mBlocked = false;
if (prevMsg != null) {
prevMsg.next = msg.next;
} else {
mMessages = msg.next;
}
msg.next = null;
if (DEBUG) Log.v(TAG, "Returning message: " + msg);
msg.markInUse();
return msg;
}
} else {
// No more messages.
nextPollTimeoutMillis = -1;
}
// Process the quit message now that all pending messages have been handled.
//检查用户是否将MessageQueue退出了。
if (mQuitting) {
//这里的销毁函数nativeDestroy(mPtr)，mPtr = 0;销毁native层的MessageQueue，并且将其在java层的指针清空。
dispose();
return null;
}
// If first time idle, then get the number of idlers to run.
// Idle handles only run if the queue is empty or if the first message
// in the queue (possibly a barrier) is due to be handled in the future.
//方法走到这里了，代表MessageQueue内暂时没有需要处理的Message对象，那么，这时如果用户设置了idlehandler的话，线程就会“忙里偷闲”抽空处理一下这个idleHandler。
if (pendingIdleHandlerCount <0
&& (mMessages == null || now <mMessages.when)) {
pendingIdleHandlerCount = mIdleHandlers.size();
}
if (pendingIdleHandlerCount <= 0) {
// No idle handlers to run. Loop and wait some more.
mBlocked = true;
continue;
}
if (mPendingIdleHandlers == null) {
mPendingIdleHandlers = new IdleHandler[Math.max(pendingIdleHandlerCount, 4)];
}
mPendingIdleHandlers = mIdleHandlers.toArray(mPendingIdleHandlers);
}
//到这里表示用户却是设置了IdleHandler，下面就是遍历列表，挨个取出处理，并把处理完的丢掉。
// Run the idle handlers.
// We only ever reach this code block during the first iteration.
for (int i = 0; i < pendingIdleHandlerCount; i++) {
final IdleHandler idler = mPendingIdleHandlers[i];
mPendingIdleHandlers[i] = null; // release the reference to the handler
boolean keep = false;
try {
keep = idler.queueIdle();
} catch (Throwable t) {
Log.wtf(TAG, "IdleHandler threw exception", t);
}
if (!keep) {
synchronized (this) {
mIdleHandlers.remove(idler);
}
}
}
// Reset the idle handler count to 0 so we do not run them again.
pendingIdleHandlerCount = 0;
// While calling an idle handler, a new message could have been delivered
// so go back and look again for a pending message without waiting.
nextPollTimeoutMillis = 0;
}
}3.然后是消息的删除，这个倒是没什么好说的，比较简单
void removeMessages(Handler h, int what, Object object) {
if (h == null) {
return;
}
synchronized (this) {
Message p = mMessages;
// Remove all messages at front.
while (p != null && p.target == h&& p.what == what
&& (object == null || p.obj == object)) {
Message n = p.next;
mMessages = n;
p.recycleUnchecked();
p = n;
}
// Remove all messages after front.
while (p != null) {
Message n = p.next;
if (n != null) {
if (n.target == h&& n.what == what
&& (object == null || n.obj == object)) {
Message nn = n.next;
n.recycleUnchecked();
p.next = nn;
continue;
}
}
p = n;
}
}
}最后，这一篇基本完成了，可能我的理解会有些错误，欢迎指正，下面一篇就是分析native层中的这几个函数。
private native static long nativeInit();
private native static void nativeDestroy(long ptr);
private native void nativePollOnce(long ptr, int timeoutMillis); /*non-static for callbacks*/
private native static void nativeWake(long ptr);
private native static boolean nativeIsPolling(long ptr);
private native static void nativeSetFileDescriptorEvents(long ptr, int fd, int events);Native层
1.首先是延时函数 private native void nativePollOnce(long ptr, int timeoutMillis);
源码位置(frameworks/base/core/jni/android_os_MessageQueue.cpp)
构造函数
NativeMessageQueue::NativeMessageQueue() :
mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {
mLooper = Looper::getForThread();
if (mLooper == NULL) {
mLooper = new Looper(false);
Looper::setForThread(mLooper);
}
}获得了native层的Looper对象
nativePollOnce
static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj,
jlong ptr, jint timeoutMillis) {
//还记得java层的mPtr么，在java层里它是Long型，在这里被强转成了NativeMessageQueue指针
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
//继续往下看
nativeMessageQueue->pollOnce(env, obj, timeoutMillis);
}void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) {
mPollEnv = env;
mPollObj = pollObj;
mLooper->pollOnce(timeoutMillis);
mPollObj = NULL;
mPollEnv = NULL;
if (mExceptionObj) {
env->Throw(mExceptionObj);
env->DeleteLocalRef(mExceptionObj);
mExceptionObj = NULL;
}
}Looper.cpp (system/core/libutils/Looper.cpp 注意6.0源码位置已变)
看这个之前需要了解一下epoll模型，简单的说，epoll使用了io中断作为监听事件发生的驱动。
int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {
int result = 0;
//循环体，不停的等待发送端口的事件，有event则取出来处理
for (;;) {
//这里的顺序非常有意思，为什么是先处理上次取出来的event，再去等待取下一个event呢？很简单，这样可以充分利用
//等待下一次event的时间
while (mResponseIndex < mResponses.size()) {
const Response& response = mResponses.itemAt(mResponseIndex++);
int ident = response.request.ident;
if (ident >= 0) {
int fd = response.request.fd;
int events = response.events;
void* data = response.request.data;
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - returning signalled identifier %d: "
"fd=%d, events=0x%x, data=%p",
this, ident, fd, events, data);
#endif
if (outFd != NULL) *outFd = fd;
if (outEvents != NULL) *outEvents = events;
if (outData != NULL) *outData = data;
//取到了数据就返回了，没有则继续循环等待
return ident;
}
}
if (result != 0) {
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - returning result %d", this, result);
#endif
if (outFd != NULL) *outFd = 0;
if (outEvents != NULL) *outEvents = 0;
if (outData != NULL) *outData = NULL;
return result;
}
//处理完上次取出的event之后，开始等待发送端口事件，没有事件在这里阻塞
result = pollInner(timeoutMillis);
}
}int Looper::pollInner(int timeoutMillis) {
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - waiting: timeoutMillis=%d", this, timeoutMillis);
#endif
// Adjust the timeout based on when the next message is due.
if (timeoutMillis != 0 && mNextMessageUptime != LLONG_MAX) {
nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
int messageTimeoutMillis = toMillisecondTimeoutDelay(now, mNextMessageUptime);
if (messageTimeoutMillis >= 0
&& (timeoutMillis < 0 || messageTimeoutMillis < timeoutMillis)) {
timeoutMillis = messageTimeoutMillis;
}
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - next message in %" PRId64 "ns, adjusted timeout: timeoutMillis=%d",
this, mNextMessageUptime - now, timeoutMillis);
#endif
}
// Poll.
int result = POLL_WAKE;
mResponses.clear();
mResponseIndex = 0;
// We are about to idle.
mPolling = true;
struct epoll_event eventItems[EPOLL_MAX_EVENTS];
//这句关键，线程在这里阻塞timeoutMillis,timeoutMillis也就是对应的java层里的when-现在的时间
//这里要分两种情况，epoll_wait监听管道的发送端口，超时时间设置为timeoutMillis。
//情况1.在等待过程中，没有消息打入管道发送端，则epoll_wait函数超时返回。这里java层对应的情况是
//MessageQueue等待when时间到来的过程中，MessageQueue没有增删改相关的Message，即队列不需要唤醒。
//情况2.在等待过程中，有消息打入管道发送端，epoll_wait监听到，从接收端取出立即返回。对应java层的情况就是
//在等待when带来的是过程中，MessageQueue里加入了或者删除了相关的Message,java层经过计算，队列需要
//被唤醒,即调用了nativeWake()函数。
int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
// No longer idling.
mPolling = false;
//这里加锁保证线程安全
// Acquire lock.
mLock.lock();
// Rebuild epoll set if needed.
if (mEpollRebuildRequired) {
mEpollRebuildRequired = false;
rebuildEpollLocked();
goto Done;
}
// Check for poll error.
if (eventCount < 0) {
if (errno == EINTR) {
goto Done;
}
ALOGW("Poll failed with an unexpected error, errno=%d", errno);
result = POLL_ERROR;
goto Done;
}
// Check for poll timeout.
if (eventCount == 0) {
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - timeout", this);
#endif
result = POLL_TIMEOUT;
goto Done;
}
// Handle all events.
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ pollOnce - handling events from %d fds", this, eventCount);
#endif
//在这里遍历取出的事件进行处理
for (int i = 0; i < eventCount; i++) {
int fd = eventItems[i].data.fd;
uint32_t epollEvents = eventItems[i].events;
if (fd == mWakeEventFd) {
if (epollEvents & EPOLLIN) {
awoken();
} else {
ALOGW("Ignoring unexpected epoll events 0x%x on wake event fd.", epollEvents);
}
} else {
ssize_t requestIndex = mRequests.indexOfKey(fd);
if (requestIndex >= 0) {
int events = 0;
if (epollEvents & EPOLLIN) events |= EVENT_INPUT;
if (epollEvents & EPOLLOUT) events |= EVENT_OUTPUT;
if (epollEvents & EPOLLERR) events |= EVENT_ERROR;
if (epollEvents & EPOLLHUP) events |= EVENT_HANGUP;
pushResponse(events, mRequests.valueAt(requestIndex));
} else {
ALOGW("Ignoring unexpected epoll events 0x%x on fd %d that is "
"no longer registered.", epollEvents, fd);
}
}
}
Done: ;
// Invoke pending message callbacks.
mNextMessageUptime = LLONG_MAX;
while (mMessageEnvelopes.size() != 0) {
nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0);
if (messageEnvelope.uptime <= now) {
// Remove the envelope from the list.
// We keep a strong reference to the handler until the call to handleMessage
// finishes. Then we drop it so that the handler can be deleted *before*
// we reacquire our lock.
{ // obtain handler
sp<MessageHandler> handler = messageEnvelope.handler;
Message message = messageEnvelope.message;
mMessageEnvelopes.removeAt(0);
mSendingMessage = true;
mLock.unlock();
#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS
ALOGD("%p ~ pollOnce - sending message: handler=%p, what=%d",
this, handler.get(), message.what);
#endif
//native层的handlerMessage
handler->handleMessage(message);
} // release handler
mLock.lock();
mSendingMessage = false;
result = POLL_CALLBACK;
} else {
// The last message left at the head of the queue determines the next wakeup time.
mNextMessageUptime = messageEnvelope.uptime;
break;
}
}
// Release lock.
mLock.unlock();
// Invoke all response callbacks.
for (size_t i = 0; i < mResponses.size(); i++) {
Response& response = mResponses.editItemAt(i);
if (response.request.ident == POLL_CALLBACK) {
int fd = response.request.fd;
int events = response.events;
void* data = response.request.data;
#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS
ALOGD("%p ~ pollOnce - invoking fd event callback %p: fd=%d, events=0x%x, data=%p",
this, response.request.callback.get(), fd, events, data);
#endif
// Invoke the callback. Note that the file descriptor may be closed by
// the callback (and potentially even reused) before the function returns so
// we need to be a little careful when removing the file descriptor afterwards.
int callbackResult = response.request.callback->handleEvent(fd, events, data);
if (callbackResult == 0) {
removeFd(fd, response.request.seq);
}
// Clear the callback reference in the response structure promptly because we
// will not clear the response vector itself until the next poll.
response.request.callback.clear();
result = POLL_CALLBACK;
}
}
return result;
}最后唤醒函数nativeWake
void Looper::wake() {
#if DEBUG_POLL_AND_WAKE
ALOGD("%p ~ wake", this);
#endif
uint64_t inc = 1;
//很简单，向管道发送端写数据
ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t)));
if (nWrite != sizeof(uint64_t)) {
if (errno != EAGAIN) {
ALOGW("Could not write wake signal, errno=%d", errno);
}
}
}以上是关于[Android源代码分析]Android消息机制，Handler，Message，Looper，MessageQueue的主要内容，如果未能解决你的问题，请参考以下文章
Android Framework 分析---2消息机制Native层