Android消息循环实现原理分析
Android消息循环实现原理分析
- Android消息循环
- 消息循环的创建
- 实现原理分析
- Looper.prepare分析
- Looper.loop分析
- Handler.sendMessage分析
- 总结
- 应用
- ps中的Sys_epoll_wait
- ANR日志中的main线程栈
- 扩展
- eventfd和epoll示例
- 参考
Android消息循环
在Android中，如果一个线程有消息循环(如UI线程)，那么其他线程可以获取它的Handler对象，使用这个Handler对象发送消息到消息循环所在的线程，这个线程收到这个消息后，可以做一些操作，最典型的就是子线程执行耗时任务，然后发消息到主线程，主线程接到消息后更新UI。
Android中和消息循环相关的几个Java类如下：
1 Looper
用于循环遍历队列中的消息
2 MessageQueue
存储消息的队列，并且获取不到消息时，会阻塞等待
3 Handler
发送消息和处理消息
消息循环的创建
我们通过aosp的源码看一下，App主线程的消息循环是如何建立的。我们知道，App进程的入口函数可以认为是ActivityThread的main函数。
public static void main(String[] args) { Trace.traceBegin(Trace.TRACE_TAG_ACTIVITY_MANAGER, "ActivityThreadMain"); // CloseGuard defaults to true and can be quite spammy. We // disable it here, but selectively enable it later (via // StrictMode) on debug builds, but using DropBox, not logs. CloseGuard.setEnabled(false); Environment.initForCurrentUser(); // Set the reporter for event logging in libcore EventLogger.setReporter(new EventLoggingReporter()); // Make sure TrustedCertificateStore looks in the right place for CA certificates final File configDir = Environment.getUserConfigDirectory(UserHandle.myUserId()); TrustedCertificateStore.setDefaultUserDirectory(configDir); Process.setArgV0("" ); Looper.prepareMainLooper(); ActivityThread thread = new ActivityThread(); thread.attach(false); if (sMainThreadHandler == null) { sMainThreadHandler = thread.getHandler(); } if (false) { Looper.myLooper().setMessageLogging(new LogPrinter(Log.DEBUG, "ActivityThread")); } // End of event ActivityThreadMain. Trace.traceEnd(Trace.TRACE_TAG_ACTIVITY_MANAGER); /// M: ANR Debug Mechanism mAnrAppManager.setMessageLogger(Looper.myLooper()); Looper.loop(); throw new RuntimeException("Main thread loop unexpectedly exited"); }可以看到，在该函数中，会调用Looper.prepareMainLooper()，Looper.loop()来创建App主线程的消息循环。
但是不只是主线程才可以有消息循环，通过正确的调用Looper.prepare()和Looper.loop()，所有线程都可以创建消息循环。下面是在App的某个线程中创建消息循环的事例代码：
public class MyLooperThread extends Thread { private static final String NAME = "MyLooperThread"; private Handler mHandler; public MyLooperThread() { super(NAME); } public Handler getHandler() { return mHandler; } @Override public void run() { Looper.prepare(); mHandler = new Handler(Looper.myLooper()); Looper.loop(); }}从另一个线程中可以获取MyLooperThread的Handler对象，并且发送消息：
@Override public void onClick(View view) { if (mMyHandler == null) { mMyHandler = mMyLooperThread.getHandler(); } mMyHandler.post(new Runnable() { @Override public void run() { Log.e(TAG, "Current Thread is " + Thread.currentThread().getName()); } }); }实现原理分析
下面我们根据aosp的源码(这里我们使用aosp 8.1的代码)来分析，这个消息循环是如何实现的。
Looper.prepare分析
首先看Looper的prepare方法的实现:
public static void prepare() { prepare(true); } private static void prepare(boolean quitAllowed) { if (sThreadLocal.get() != null) { throw new RuntimeException("Only one Looper may be created per thread"); } sThreadLocal.set(new Looper(quitAllowed)); }首先prepare会创建一个Looper对象，然后将Looper对象存放到ThreadLocal中，将这个Looper对象和当前线程关联在一起。Looper的初始化在构造函数中，我们接下去看Looper的构造函数:
private Looper(boolean quitAllowed) { mQueue = new MessageQueue(quitAllowed); mThread = Thread.currentThread(); }可以看到，在Looper中会创建一个MessageQueue对象。我们接下去看MessageQueue的构造函数：
MessageQueue(boolean quitAllowed) { mQuitAllowed = quitAllowed; mPtr = nativeInit(); }MessageQueue的构造函数调用了nativeInit方法：
private native static long nativeInit();接下来继续看JNI中的实现。nativeInit方法的native实现在frameworks/base/core/jni/android_os_MessageQueue.cpp中，对应的方法是android_os_MessageQueue_nativeInit:
static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) { NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue(); if (!nativeMessageQueue) { jniThrowRuntimeException(env, "Unable to allocate native queue"); return 0; } nativeMessageQueue->incStrong(env); return reinterpret_cast<jlong>(nativeMessageQueue);}在这个方法中，创建了一个NativeMessageQueue对象，我们接下去看NativeMessageQueue的构造函数，也实现在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); }}在NativeMessageQueue中创建了一个Looper，这个Looper是native层的，实现在system/core/libutils/Looper.cpp中。接下来我们从Looper的构造函数开始：
Looper::Looper(bool allowNonCallbacks) : mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false), mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false), mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) { 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_CALLBACKS ALOGD("%p ~ rebuildEpollLocked - rebuilding epoll set", this);#endif close(mEpollFd); } // Allocate the new epoll instance and register the wake pipe. 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 eventItem.events = EPOLLIN; eventItem.data.fd = mWakeEventFd; 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)); for (size_t i = 0; i < mRequests.size(); i++) { const Request& request = mRequests.valueAt(i); struct epoll_event eventItem; request.initEventItem(&eventItem); int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, & eventItem); if (epollResult < 0) { ALOGE("Error adding epoll events for fd %d while rebuilding epoll set: %s", request.fd, strerror(errno)); } }}我们只解释最关键的地方，其他不相关的代码，先不解释。
在Looper构造方法中，首先调用mWakeEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);创建一个mWakeEventFd，这个mWakeEventFd可以认为是一个用于进程/线程间通信的文件，不是真实的磁盘文件，而是linux内核的虚拟文件系统(VFS)抽象出来的，它的底层只是对应某个数据结构，该数据结构里有一些字段来标示它的状态，具体实现细节我们这里不深究。
然后调用rebuildEpollLocked函数，该函数中，调用mEpollFd = epoll_create(EPOLL_SIZE_HINT);创建一个mEpollFd，这个mEpollFd代表一个epoll实例，epoll实例用于支持高效的IO多路复用，这里我们先不管什么是IO多路复用，只需要知道它可以阻塞监听一个或多个文件，当被监听的文件发生变化的时候，会被唤醒。
最后调用epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);将上面创建的mWakeEventFd交给epoll来管理。可以认为使用epoll来监听mWakeEventFd，当有人向mWakeEventFd中写入数据的时候，epoll会被唤醒。
Looper.loop分析
接下来我们分析Looper.loop这条线(这里指的是Java层的Looper)。Looper中loop的实现如下：
public static void loop() { final Looper me = myLooper(); if (me == null) { throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread."); } 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. Binder.clearCallingIdentity(); final long ident = Binder.clearCallingIdentity(); for (;;) { Message msg = queue.next(); // might block if (msg == null) { // No message indicates that the message queue is quitting. return; } ...... ......该方法调用MessageQueue的next方法去获取消息，我们知道next方法是会睡眠的，如果MessageQueue里面没有消息，那么会next方法会一直等待，直到有一个新的消息发到了MessageQueue中，当前线程会被唤醒，然后next方法继续执行，并返回这个新的消息(这里我们不考虑延迟消息这种情况)。我们接下去继续分析next方法，看它是如何实现睡眠等待的。
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. 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.flushPendingCommands(); } nativePollOnce(ptr, nextPollTimeoutMillis); 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. do { prevMsg = msg; msg = msg.next; } while (msg != null && !msg.isAsynchronous()); } if (msg != null) { if (now < msg.when) { // 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. ...... ......关键的地方是调用nativePollOnce:
private native void nativePollOnce(long ptr, int timeoutMillis); /*non-static for callbacks*/对应的native实现是frameworks/base/core/jni/android_os_MessageQueue.cpp中的android_os_MessageQueue_nativePollOnce函数:
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);}调到nativeMessageQueue的pollOnce，该函数同样实现在frameworks/base/core/jni/android_os_MessageQueue.cpp中:
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; }}这里调用了native层的Looper的pollOnce函数，该函数实现在system/core/libutils/Looper.cpp中:
int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) { int result = 0; for (;;) { 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; } result = pollInner(timeoutMillis); }}我们略过一些不重要的逻辑，直接看pollInner函数:
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]; int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis); // No longer idling. mPolling = false; ...... ...... 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 { ...... ......在这里，关键的地方是epoll_wait的调用和awoken的调用。
在pollInner里，调用epoll_wait阻塞监听上一步使用eventfd()创建的mWakeEventFd，只要有人向mWakeEventFd里写入一个64位的无符号整数(uint64_t)，epoll_wait就会被唤醒，并且将唤醒事件填充到eventItems数组中，然后通过for循环检查epoll_wait返回的事件，如果是mWakeEventFd的EPOLLIN事件，则说明有人向mWakeEventFd中写入了数据。然后调用awoken读出mWakeEventFd中的数据。我们看一下awoken的实现:
void Looper::awoken() {#if DEBUG_POLL_AND_WAKE ALOGD("%p ~ awoken", this);#endif uint64_t counter; TEMP_FAILURE_RETRY(read(mWakeEventFd, &counter, sizeof(uint64_t)));}可以看到，awoken确实只是读出mWakeEventFd中的数据而已，并且读出的数据也没什么用。
既然已经唤醒了，为什么还要读出mWakeEventFd中的数据呢？这个数据并没有什么意义。这是由于epoll的触发模式决定的。
epoll的触发模式分为两种：水平触发和边缘触发。水平触发的意思是只要被监听的fd中存在数据，epoll_wait就一直返回，只有将fd中的数据读出，才会再次阻塞，等待下次写入再唤醒。边缘触发是只有向fd中写入数据，才会触发epoll_wait返回，epoll_wait返回后，不必读出fd中的数据，因为不管fd中是否有数据，epoll_wait都会再次等待有人写入。
在Looper.cpp中，使用的是水平触发，所以需要读出里面的数据，epoll_wait才会正常等待下一次事件。
是水平触发还是边缘触发，是在通过epoll_ctl监听文件的时候指定的，通过struct epoll_event的events字段指定。默认是水平触发，只需要指定eventItem.events = EPOLLIN，下面是Looper.cpp中通过epoll监听mWakeEventFd的代码:
struct epoll_event eventItem; memset(& eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union eventItem.events = EPOLLIN; eventItem.data.fd = mWakeEventFd; int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);如果采用边缘触发模式，则需要指定EPOLLET:
eventItem.events = EPOLLIN | EPOLLET;Handler.sendMessage分析
当执行完Looper.prepare()后，就可以获取线程的handler对象，通过handler对象向消息队列中发送消息，消息的处理就会在当前线程中处理。下面看一下，为什么发送一个消息，就能唤醒当前线程，并且在当前线程处理消息。
先看Handler的sendMessage方法(post也会调用sendMessage方法)：
Handler.java
public final boolean sendMessage(Message msg) { return sendMessageDelayed(msg, 0); } public final boolean sendMessageDelayed(Message msg, long delayMillis) { if (delayMillis < 0) { delayMillis = 0; } return sendMessageAtTime(msg, SystemClock.uptimeMillis() + delayMillis); } 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.java
boolean enqueueMessage(Message msg, long when) { if (msg.target == null) { throw new IllegalArgumentException("Message must have a target."); } if (msg.isInUse()) { throw new IllegalStateException(msg + " This message is already in use."); } synchronized (this) { 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.markInUse(); msg.when = when; Message p = mMessages; boolean needWake; 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. needWake = mBlocked && p.target == null && msg.isAsynchronous(); 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; } // We can assume mPtr != 0 because mQuitting is false. if (needWake) { nativeWake(mPtr); } } return true; }private native static void nativeWake(long ptr);Handler的sendMessage一直会调到MessageQueue的enqueueMessage中，在equeueMessage中，先将消息放到队列上，然后调用nativeWake来唤醒队列。
nativeWake对应的jni方法，实现在frameworks/base/core/jni/android_os_MessageQueue.cpp中：
static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) { NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr); nativeMessageQueue->wake();}调用了NativeMessageQueue的wake方法，该方法也实现在frameworks/base/core/jni/android_os_MessageQueue.cpp中:
void NativeMessageQueue::wake() { mLooper->wake();}继续调用到Looper的wake方法，下面继续看Looper的wake方法，该方法实现在system/core/libutils/Looper.cpp中：
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) { LOG_ALWAYS_FATAL("Could not write wake signal to fd %d: %s", mWakeEventFd, strerror(errno)); } }}Looper的wake方法中，向mWakeEventFd写入一个uint64_t。前面分析Looper.prepare，Looper.loop时讲到，mWakeEventFd是通过eventfd()创建的一个专门用于进程(或线程)之间通知的文件，并且该文件通过epoll监控，所以这里写入一个数据，就可以唤醒epoll_wait，也就是唤醒消息循环线程，消息循环线程被唤醒后，就会取队列中的消息去执行，如果队列中没有消息，又会继续调到epoll_wait进行等待。当下一次有人向mWakeEventFd写数据，也就是在java层中有人使用handler发送消息，则该线程又会被唤醒去执行消息。
总结
经过上面的分析，我们可以知道，Android的消息循环的基础是epoll监控eventfd，对应关系如下:
Looper.prepare对应eventfd()，epoll_create，epoll_ctl
Looper.loop 对应epoll_wait
Handler.sendMessage对应write eventfd
应用
其实即使我们不深入源码分析，我们平时做开发，也可以看到epoll的存在，只是可能没有深入去思考它到底是什么。下面就列举一下平时看到的epoll:
ps中的Sys_epoll_wait
在使用ps打印进程信息时，会有一列叫做WCHAN，WCHAN的意义是，当前进程正在哪个内核函数上等待。
可以看到，这个进程的WCHAN为Sys_epoll_，其实这里没显示完整，完整的WCHAN值为Sys_epoll_wait，可以通过查看/proc/pid/wchan来查看：
epoll_wait对应的系统调用就是Sys_epoll_wait，所以可以确认，当前进程正在等待epoll_wait返回，因为消息循环是通过epoll实现的，所以得到的结论就是：这个进程正在消息队列上等待。
ANR日志中的main线程栈
在分析anr的时候，经常看到主线程或者带有消息循环的线程的状态是这样的：
这种状态也是在消息队列上等待。调用栈如下，
Looper.loop
MessageQueue.next
MessageQueue.nativePollOnce
android_os_MessageQueue_nativePollOnce
android::NativeMessageQueue.pollOnce
android::Looper.pollOnce
android::Looper.pollInner
epoll_wait
和我们之前分析源码的调用流程是一致的。
扩展
epoll是一种IO多路复用技术，也就是通过一个线程可以监控多个fd。像一些普通的系统调用，如read，只能监控一个fd，如果要监控多个fd,就要启动多个线程，如果需要监控的fd比较多，如大型服务器中的socket很多，如果每个socket都起一个线程来监控，系统中会出现大量线程，对系统造成很大的压力。
而epoll使用一个线程，就能监控多个fd。大大降低了系统中的线程数量。
如何用epoll监控多个fd呢？只需要调用epoll_ctl将fd加到它的监控集合中即可。如下:
int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, & eventItem);c++层的Looper是支持监控多个fd的：
int Looper::addFd(int fd, int ident, int events, Looper_callbackFunc callback, void* data) { return addFd(fd, ident, events, callback ? new SimpleLooperCallback(callback) : NULL, data);}int Looper::removeFd(int fd) { return removeFd(fd, -1);}其实addFd的实现就是调用epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, & eventItem);，removeFd的实现就是调用epoll_ctl(mEpollFd, EPOLL_CTL_DEL, fd, NULL)。
而Android的消息循环，只监控了一个mWakeEventFd，所以Android的消息循环只使用了Looper.cpp的部分功能。
eventfd和epoll示例
使用c++写一个用epoll监控eventfd的示例，代码量很小：
#include 参考
eventfd：
http://man7.org/linux/man-pages/man2/eventfd.2.html
epoll:
http://man7.org/linux/man-pages/man7/epoll.7.html
更多相关文章
- C语言函数以及函数的使用
- Android架构分析之Android消息处理机制（三）
- Android 安全加密：消息摘要Message Digest详解
- Android架构分析之Android消息处理机制（一）
- 深入理解 Android消息处理系统的原理
- Android UI线程和非UI线程
- 【Android自助餐】Handler消息机制完全解析（三）Handler解析
- Android进程间通信--消息机制及IPC机制实现