Android Handler消息队列的实现原理

我们在写Android程序的时候，有经常用到Handler来与子线程通信，亦或者是用其来管理程序运行的状态时序。Handler其是由Android提供的一套完善的操作消息队列的API。它既可以运行在主线程中，也可以运行在子线程中，唯一的区别是其内部的Looper对象不同。在这里我将对Android中Handler消息队列的实现进行一个总结，以便深入了解其原理并且巩固其使用方式。

本系列的主要内容如下：

1.整体架构与主要的数据结构
2.Message的入列出列以及延迟处理的实现方式
3.底层原理：epoll的基础概念以及在Handler中的应用

整体架构

实现Handler消息队列的源码如下：

framework/base/core/java/android/os/Handler.java
framework/base/core/java/android/os/Looper.java
framework/base/core/java/android/os/Message.java
framework/base/core/java/android/os/MessageQueue.java
framework/base/core/jni/android_os_MessageQueue.cpp
system/core/libutils/Looper.cpp
system/core/include/utils/Looper.h

Handler: 主要提供各类API用于将Message对象加入到队列中，并在Message从队列中被取出时，触发handleMessage回调或者执行Message中的Runable对象的run函数。

Looper: java层的Looper类是消息队列的工作引擎，提供一个死循环不断从MessageQueue中取出Message, 取出的Message会在Handler的handleMessage中处理。

Message: 消息队列中所传递的消息的数据类型，其父类是Parcelable。内部还包含了一个Message对象池的实现，用来复用Message对象；

MessageQueue：消息队列，主要实现Message的入列，出列的逻辑管理，维护Message单链表，队列的清空，以及与JNI部分的通信。

android_os_MessageQueue: 消息队列的JNI部分实现，内部的主要逻辑就是初始化native的Looper对象，以及在java部分调用JNI中的方法时，执行native Looper部分对应的方法。

native Looper: native层的Looper中实现了两个功能，一个是利用epoll实现了队列的阻塞与唤醒功能，另外一个是实现了一套native层中使用的队列机制。

其分层架构如下所示：Handler, Message，Looper三者围绕MessageQueue进行处理，MessageQueue通过jni与Looper.cpp进行通信实现空队列时阻塞，有消息入列时唤醒等功能。

整体架构如下：MessageQueue中维护着一个Message的单链表，Handler中enqueueMessage将消息添加到队列中，Looper.java从队列中取出Message并将其交由Handler的dispatchMessage分发。android_os_MessageQueue.cpp主要是jni的实现部分，完成MessageQueue跟Looper.cpp的交互。 Looper.cpp则使用epoll机制实现了队列的阻塞与唤醒。

数据结构

Android的队列所管理的数据类型为Message, 消息队列所用的数据结构则是Message类型的单链表。一般模式下其结构如下所示：

在Looper中，依次从Message 0 到 Message n中将数据取出丢给Handler处理。当添加新的数据时，如果Message对象的when属性为0，则将其添加到链头。这种情况一般是我们调用Handler的sendMessageAtFrontOfQueue时出现。

在其他情况下，每一次添加新的消息时，都需要从链头依次对比Message的when变量，当新消息的when变量小于链中元素的when变量，则将其插入链中。一般调用sendEmptyMessageAtTime，sendEmptyMessageDelayed sendMessageDelayed，sendMessageAtTime等函数时，会出现延迟处理的消息。如果队列中存在延迟消息，那么使用sendMessage等及时处理的消息时，会出现其插入在链表中的情况。

以上总结了队列的数据结构是一个单链表，因此我们上述所有疑问，都可以转换成如何操作单链表的问题。正常情况下的单链表如下所示：

那Message究竟是怎么加入到队列中，又是怎么从队列中取出最终到handlerMessage中为我们所得，为什么队列可以将数据优先处理，为什么有的数据可以延迟处理，其具体实现队列管理的算法是怎么实现的。

首先看一下，Handler是如何从队列中取值的，具体实现在MessageQueue中的next()函数

//获取当前时间戳
final long now = SystemClock.uptimeMillis();
Message prevMsg = null;
//mMessages是一个全局的Message类型变量，保存着链头的Message, 如果队列没有数据，则该变量指向空
Message msg = mMessages;
if (msg != null && msg.target == null) {//无主的Message，不处理，直接寻找下一个节点的Message.// Stalled by a barrier.  Find the next asynchronous message in the queue.do {prevMsg = msg;msg = msg.next;} while (msg != null && !msg.isAsynchronous());}if (msg != null) {//队列中有数据需要处理if (now < msg.when) {//当前时间还没达到队列中第一个Message的消息处理时间，需要继续等待。// Next message is not ready.  Set a timeout to wake up when it is ready.nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);} else {// Got a message.mBlocked = false;//返回当前链头的Message,并将mMessages对象指向其内部名为next的Message类型的对象。//这里就相当于取出了队列的头部数据。if (prevMsg != null) {prevMsg.next = msg.next;} else {mMessages = msg.next;}msg.next = null;if (DEBUG) Log.v(TAG, "Returning message: " + msg);//标识当前Message正在使用中msg.markInUse();//返回Message对象。return msg;}
} else {// No more messages.nextPollTimeoutMillis = -1;
}

这部分仅仅列出取值的具体算法，其他包括空队列的IdleHandler的处理，以及阻塞的逻辑的部分这里我并没有贴出来。

Looper调用MessageQueue对象的next()方法获取到从队列中返回的Message对象后，再将其交由Handler去处理，也就是走到我们经常用到的handleMessage(Message msg)回调中，这部分的逻辑如下：

Message msg = queue.next(); // might block
if (msg == null) {// No message indicates that the message queue is quitting.return;}......
......try {msg.target.dispatchMessage(msg);
} finally {if (traceTag != 0) {Trace.traceEnd(traceTag);}
}

每一个Message对象都有一个target变量，这个target变量类型就是Handler, 每当Message取出时，都提交到其自身归属的Handler去处理。

每次取消息的时序图如上所示，MessageQueue由Looper初始化，初始化之后，调用其loop方法开始让队列开始工作。loop方法中，开始循环调用MessageQueue的next()函数去拿Message数据，next()具有阻塞特性，当队列没有消息时，nativePollOnce()函数处会阻塞，原理是底层使用epoll实现了消息的阻塞/唤醒机制。当队列中有新加入的数据时，epoll_wait就会退出，从而MessageQueue的next()方法将Message返回给Looper,并提交到Handler中消化。

知道了Handler的消息取出的流程，接下来看一下将消息加入队列的逻辑。Handler一共为我们提供了如下公开的API用来将消息添加到队列中

//发送一个空消息public final boolean sendEmptyMessage(int what)//发送一个延迟处理的空消息public final boolean sendEmptyMessageDelayed(int what, long delayMillis)//发送一个指定时间戳执行的空消息public final boolean sendEmptyMessageAtTime(int what, long uptimeMillis)//发送一个延迟消息public final boolean sendMessageDelayed(Message msg, long delayMillis)//发送一个定时消息public boolean sendMessageAtTime(Message msg, long uptimeMillis)//将消息发送至队列的头public final boolean sendMessageAtFrontOfQueue(Message msg)//提交一个任务public final boolean post(Runnable r)//提交一个定时任务public final boolean postAtTime(Runnable r, long uptimeMillis)//提交一个带身份认证的定时任务public final boolean postAtTime(Runnable r, Object token, long uptimeMillis)//提交一个延迟任务public final boolean postDelayed(Runnable r, long delayMillis)//提交一个任务到队列头public final boolean postAtFrontOfQueue(Runnable r)

以上方法，最终均会调用Handler的私有方法将Message提交到队列中，uptimeMillis就是该msg最终执行的时间戳，也用来进行链中元素的排序。

private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {msg.target = this;if (mAsynchronous) {msg.setAsynchronous(true);}return queue.enqueueMessage(msg, uptimeMillis);}

接下来看一下消息插入的具体实现，MessageQueue中的enqueueMessage方法。

boolean enqueueMessage(Message msg, long when) {......//标记当前Message正在被使用msg.markInUse();//将Message的when变量赋值为执行时间戳msg.when = when;//mMessages为链头，如果此时队列无数据，则mMessages为空Message p = mMessages;boolean needWake;if (p == null || when == 0 || when < p.when) {//这里判断有三种情况，一种是链头为空表示当前队列无数据，一种是当前提交的Message执行时间戳为0，表示使用者//希望将其加入到队列头，还有一种情况是当前提交的Message执行时间戳小于链头Message的执行时间戳，因此也需    //要将其加入到队列头                // New head, wake up the event queue if blocked.//加入队列头只需要将自身的next变量引用到当前的链头对象，然后再将代表当前链头的mMessages变量引用到msgmsg.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.needWake = mBlocked && p.target == null && msg.isAsynchronous();//开始遍历链表中的每个元素，找到新Message的插入位置。Message的插入位置//是根据其内部的when变量来判断，当链表中的元素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.nextprev.next = msg;}// We can assume mPtr != 0 because mQuitting is false.//队列为空时，此时epoll_wait处于等待状态，我们需要将其唤醒，然后loop可以继续从队列中取出数据，分发数据。if (needWake) {nativeWake(mPtr);}}return true;}

运行时序如下所示，该图描述了一个Message数据加入空队列，到数据被取出消化的流程。

总结：Android消息队列中的Message入列和出列，都是基于单链表来实现，其队列排序的核心变量就是Message内部的when变量，when变量是一个时间戳，由Handler给该变量赋值，延迟消息，定时消息，都是根据when变量来实现。在这里分析队列的逻辑时，发现了跟jni部分的通信，主要是nativePollOnce，nativeWake方法，这两个方法实际是实现了空队列阻塞，以及唤醒的功能，底层使用epoll机制实现。

上述分析中讲了Handler的消息入列与出列的具体实现时，其中有碰到MessageQueue中使用了两个jni的函数，nativePollOnce和nativeWake，nativePollOnce的作用一个是保证了Looper循环在消息队列中没有数据或者链头的Message的when变量大于当前时间戳时能够阻塞，从而减少CPU的资源使用率，nativeWake的作用则是在Looper循环被阻塞的时候，当有新的消息加入到队列中执行时，能够及时唤醒阻塞的循环，保证消息能够及时处理。那这两个函数是如何实现阻塞/唤醒的呢？首先跟踪两者的本地函数定义在MessageQueue中：

 private native void nativePollOnce(long ptr, int timeoutMillis); /*non-static for callbacks*/private native static void nativeWake(long ptr);

对应的JNI函数在android_os_MessageQueue.cpp中，如下：

static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj,jlong ptr, jint timeoutMillis) {NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);nativeMessageQueue->pollOnce(env, obj, timeoutMillis);
}static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);nativeMessageQueue->wake();
}

jni函数中并没有做处理，只是调用NativeMessageQueue的方法，继续跟踪两个函数在NativeMessageQueue的执行如下：

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;}
}void NativeMessageQueue::wake() {mLooper->wake();
}

发现最终的实现都是在Looper.cpp中，在分析Looper.cpp的源码时，有必要了解一下epoll的基础概念，以及使用方式，参考这篇文章。
native Looper对象的初始化，在android_os_Message.cpp文件的NativeMessageQueue构造函数中完成

NativeMessageQueue::NativeMessageQueue() :mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {//从当前线程的私有域中查看是否有已经创建的LoopermLooper = Looper::getForThread();if (mLooper == NULL) {//如果没有已经存在的Looper,则创建新的mLooper = new Looper(false);Looper::setForThread(mLooper);}
}

继续跟踪Looper的构造函数

Looper::Looper(bool allowNonCallbacks) :mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false),mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {//eventfd 是 Linux 的一个系统调用，创建一个文件描述符用于事件通知//具体参数介绍以及使用方式，请看[这里](http://man7.org/linux/man-pages/man2/eventfd2.2.html)mWakeEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);LOG_ALWAYS_FATAL_IF(mWakeEventFd < 0, "Could not make wake event fd: %s", strerror(errno));AutoMutex _l(mLock);rebuildEpollLocked();
}void Looper::rebuildEpollLocked() {// Close old epoll instance if we have one.if (mEpollFd >= 0) {
#if DEBUG_CALLBACKSALOGD("%p ~ rebuildEpollLocked - rebuilding epoll set", this);
#endifclose(mEpollFd);}// Allocate the new epoll instance and register the wake pipe.//mEpollFd是epoll创建的文件描述符mEpollFd = epoll_create(EPOLL_SIZE_HINT);LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance: %s", strerror(errno));struct epoll_event eventItem;memset(& eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union//标记mWakeEventFd对read操作有效eventItem.events = EPOLLIN;eventItem.data.fd = mWakeEventFd;//epoll_ctl执行EPOLL_CTL_ADD参数的操作的意思是将mWakeEventFd加入到监听链表中，当有read操作时，唤醒mWakeEventFd的wait等待。int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);LOG_ALWAYS_FATAL_IF(result != 0, "Could not add wake event fd to epoll instance: %s",strerror(errno));............

Looper的构造函数中，通过epoll_create获得了一个epoll的文件描述符，再通过epoll_ctl将mWakeEventFd添加到epoll中监听。从上面个跟踪的流程得知，我们调用nativePollOnce()函数，最终执行的地方是在Looper的pollOnce中，代码如下：

int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {int result = 0;for (;;) {............if (result != 0) {
#if DEBUG_POLL_AND_WAKEALOGD("%p ~ pollOnce - returning result %d", this, result);
#endifif (outFd != NULL) *outFd = 0;if (outEvents != NULL) *outEvents = 0;if (outData != NULL) *outData = NULL;return result;}result = pollInner(timeoutMillis);}
}int Looper::pollInner(int timeoutMillis) {
#if DEBUG_POLL_AND_WAKEALOGD("%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_WAKEALOGD("%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;//开始等待事件的触发， timeoutMillis是在该时间内，如果没有获取到事件，则自动返回，为-1则一直等待到有事件过来，为0则不管有没有事件，都    //立即返回。 Handler在设计的时候，首先会查询一次队列，如果没有数据，则立即返回，然后重新pollOnce走到这里等待新的数据过来， 往    // mWakeEventFd写数据，才会唤醒返回。struct epoll_event eventItems[EPOLL_MAX_EVENTS];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: %s", strerror(errno));result = POLL_ERROR;goto Done;}//请求超时// Check for poll timeout.if (eventCount == 0) {
#if DEBUG_POLL_AND_WAKEALOGD("%p ~ pollOnce - timeout", this);
#endifresult = POLL_TIMEOUT;goto Done;}// Handle all events.
#if DEBUG_POLL_AND_WAKEALOGD("%p ~ pollOnce - handling events from %d fds", this, eventCount);
#endif
......
......

在队列消息为空的情况下，那么我们就会阻塞在poll_inner的epoll_wait处，如果有新消息加入队列，则上层会调用nativeWake，最终对应Looper中的wake函数将epoll_wait唤醒。


void Looper::wake() {
#if DEBUG_POLL_AND_WAKEALOGD("%p ~ wake", this);
#endifuint64_t inc = 1;//往mWakeEventFd中写数据， mWakeEventFd监听到有事件触发，则使epoll_wait返回。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: %s", strerror(errno));}}
}

至此，Looper.cpp中实现队列的阻塞唤醒的功能已经完成，总结一下：

Looper.cpp对象是在android_os_MessageQueue的nativeInit函数被调用时初始化
Looper.cpp的构造函数中，使用eventfd函数创建了一个mWakeEventFd的文件描述符用于事件通知，并使用epoll_create函数创建了一个epoll的文件描述符。然后调用epoll_ctl将mWakeEventFd加入到了事件监听链中。
初始化的队列为空，请求数据的时候，首先会走到pollOnce中，传入的timeoutMillis值为0，则epoll_wait会立即返回，然后MessageQueueh会进行timeoutMills = -1的第二次pollOnce, 这时候就阻塞在epoll_wait函数处。
当有新数据加入队列时，调用wake函数往mWakeEventFd中写数据, 此时触发epoll_wait阻塞中断，pollOnce返回，上层MessageQueue中的next()函数也将执行完毕，并将Message丢给Handler处理