Android Looper与Message
Android的Handler机制是一个非常强大的消息的机制系统,Handler机制的使用在Android整个系统中随处可见。
Handler作为消息中转器和处理器,是用户进行消息处理的最直接接口,而消息的队列处理和按序分发,即Handler机制的背后支持,则是由Looper和MessageQueue来进行处理。
这里不对Handler做相关介绍,重点是Looper和MessageQueue怎么来配合完成的。
从源码角度看从 handler.sendMessage 到 handleMessage 的整个过程。
以Handler作为入口:
sendMessage:
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; } //这里开始进入MessageQueue逻辑 return enqueueMessage(queue, msg, uptimeMillis); }
MessageQueue
调用MessageQueue的 enqueueMessage
/* * 1. 非法情况判别 * 2. 当队列为空,msg入队,唤醒epoll_wait * 3. 队列不为空,链表末尾,加入msg * */ 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("MessageQueue", 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) { //当队列还没有任何消息的时候 //将message入队,这里是链表的形式, 并没有队列或者数组来进行维护 // New head, wake up the event queue if blocked. msg.next = p; mMessages = msg; //当队列为空,则looper会处于block状态 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加入到队列末尾 msg.next = p; // invariant: p == prev.next prev.next = msg; } // We can assume mPtr != 0 because mQuitting is false. if (needWake) { //这里是唤醒底层的epoll_wait,此时可能looper正处于空队列等待状态,有新加入的message后需要唤醒 nativeWake(mPtr); } } return true; }
这里有几个重要的变量:
// True if the message queue can be quit. //是否可以被退出 private final boolean mQuitAllowed;
//NativeMessageQueue对象的地址,MessageQueue的等待和唤醒机制是在JNI下执行的 @SuppressWarnings("unused") private long mPtr; // used by native code //Message链表的头 Message mMessages; //IdleHandler集合,用来在队列执行到空的时候进行回调 private final ArrayList<IdleHandler> mIdleHandlers = new ArrayList<IdleHandler>(); //mIdleHandlers的数组形式,在调用的时候会对mIdleHandlers进行一份拷贝 private IdleHandler[] mPendingIdleHandlers; //是否MessageQueue正在退出(退出操作使用Looper.quit进行触发) private boolean mQuitting; // Indicates whether next() is blocked waiting in pollOnce() with a non-zero timeout. //标记当前Lopper是否处于block状态 private boolean mBlocked; // The next barrier token. // Barriers are indicated by messages with a null target whose arg1 field carries the token. private int mNextBarrierToken; private native static long nativeInit(); private native static void nativeDestroy(long ptr); //nativePollOnce主要触发等待机制 private native static void nativePollOnce(long ptr, int timeoutMillis); //nativeWake在合适是时机下唤醒epoll_wait等待 private native static void nativeWake(long ptr); private native static boolean nativeIsIdling(long ptr);
入队之后, Looper会被唤醒, 并从MessageQueue中依次取出message,直到MessageQueue为空(所有的消息都被处理完了);
Looper
Looper对象是每一个线程有一个,这个是一一对应的关系,每一个Looper会有一个MessageQueue, 同样是一一对应关系。
Looper.prepare
public static void prepare() { prepare(true); } //quitAllowed对应MessageQueue中的mQuitAllowed, 在创建Looper的时候就已经指定, 默认可以被取消,否则会抛出异常 //这里没有提供对应的接口,所以所有的Looper理论上都是可以被取消的
private static void prepare(boolean quitAllowed) { if (sThreadLocal.get() != null) { throw new RuntimeException("Only one Looper may be created per thread"); } //用ThreadLocal为当前线程唯一保存一份Looper对象 sThreadLocal.set(new Looper(quitAllowed)); }
每个looper对应一个MessageQueue:
private Looper(boolean quitAllowed) { mQueue = new MessageQueue(quitAllowed); mThread = Thread.currentThread(); }
Looper.loop()
/* * 1. 从MessageQueue中取出下一个待处理的Message对象; * 2. 调用对应的Handler的dispatchMessage分发消息 * */ 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 (;;) { //取得MessageQueue中的下一个待处理的Message Message msg = queue.next(); // might block 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); } //分发消息,target是msg中绑定的Handler,这里就会发送msg对象到handleMessage中进行处理. msg.target.dispatchMessage(msg); if (logging != null) { logging.println("<<<<< Finished to " + msg.target + " " + msg.callback); } // 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); } msg.recycleUnchecked(); } }
Handler.dispatchMessage:
public void dispatchMessage(Message msg) { if (msg.callback != null) { //首先优先调用Message的callback(Runnable)的run方法 handleCallback(msg); } else { if (mCallback != null) { //如果essage的callback(Runnable)不存在,其次调用Handler的CallBack对象的handleMessage if (mCallback.handleMessage(msg)) { return; } } //最后才会调用Handler自己的handleMessage方法 handleMessage(msg); } }
最终会走到我们的handleMessage中, 然后根据情况进行处理.
然而这里的重点,其实是Looper.loop中的queue.next()方法,注意这里的注释(该方法可能会阻塞):
Message msg = queue.next(); // might block
MessageQueue.next()
/* * 1. 循环从队列中取Message; * 2. 判断该Message是否是一个Barrier, 如果是,则不执行后面所有的同步的message,异步Message不受影响, 默认的message都是同步的; * 3. 如果没有Barrier, 则依次从Message链表中取出前面的Message; * 4. 判断当前的MessageQueue是否有设置IdleHandler,在队列为空的时候去回调他的方法queueIdle. * */ 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; } //标记IdleHandler的数量 int pendingIdleHandlerCount = -1; // -1 only during first iteration //下次epoll_wait的超时时间, 0会立刻返回, -1 则会永远等待, >0即设置的超时等待时间 int nextPollTimeoutMillis = 0; for (;;) { if (nextPollTimeoutMillis != 0) { Binder.flushPendingCommands(); } //触发epoll_wait机制 nativePollOnce(ptr, nextPollTimeoutMillis); synchronized (this) { // Try to retrieve the next message. Return if found. final long now = SystemClock.uptimeMillis(); //从队列中取出message Message prevMsg = null; Message msg = mMessages; //判断是否是一个barrier[MessageQueue 的Barrier后面会说到] if (msg != null && msg.target == null) { // Stalled by a barrier. Find the next asynchronous message in the queue. //这里查找被Barrier绊住之后的所有非异步消息, 异步消息不会受到barrier的影响. //同时这里会把msg的指针移到最后(null), 目的是让barrier之后的所有同步(默认为同步)消息都没办法执行. do { prevMsg = msg; msg = msg.next; } while (msg != null && !msg.isAsynchronous()); } //如果有barrier, 这里不会走 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. //从链表中拿出一个msg mBlocked = false; if (prevMsg != null) { prevMsg.next = msg.next; } else { mMessages = msg.next; } //将message从链表中剥离 msg.next = null; if (false) Log.v("MessageQueue", "Returning message: " + msg); return msg; } } else { // No more messages. //epoll_wait会一直等待 nextPollTimeoutMillis = -1; } // Process the quit message now that all pending messages have been handled. if (mQuitting) { 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. 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; } //拿到所有IdleHandler拷贝的数组形式 if (mPendingIdleHandlers == null) { mPendingIdleHandlers = new IdleHandler[Math.max(pendingIdleHandlerCount, 4)]; } mPendingIdleHandlers = mIdleHandlers.toArray(mPendingIdleHandlers); } // 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 { // 当队列为空的时候,回调queueIdle, 返回值决定该IdleHandler是在执行之后被保留; // 注意: 不保留只会在MessageQueue首次为空的时候被调用, 保留则每次为空都会被调用 // 如果当前一直为空而没有任何消息,则在保留的情况下,该方法会一直被循环回调,这时候epoll_wait会直接返回而不会等待. keep = idler.queueIdle(); } catch (Throwable t) { Log.wtf("MessageQueue", "IdleHandler threw exception", t); } //如果不保留会从集合中移除该IdleHandler. 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. //回调IdleHandler也存在有新的message消息被发送的可能性,所以这里不会阻塞. nextPollTimeoutMillis = 0; } }
nativePollOnce
core/jni/android_os_MessageQueue.cpp
tatic JNINativeMethod gMessageQueueMethods[] = { /* name, signature, funcPtr */ { "nativeInit", "()J", (void*)android_os_MessageQueue_nativeInit }, { "nativeDestroy", "(J)V", (void*)android_os_MessageQueue_nativeDestroy }, { "nativePollOnce", "(JI)V", (void*)android_os_MessageQueue_nativePollOnce }, { "nativeWake", "(J)V", (void*)android_os_MessageQueue_nativeWake }, { "nativeIsIdling", "(J)Z", (void*)android_os_MessageQueue_nativeIsIdling } };
对应的NativeMessageQueue的pollOnce:
static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jclass clazz, jlong ptr, jint timeoutMillis) { NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr); nativeMessageQueue->pollOnce(env, timeoutMillis); }
NativeMessageQueue的指针对象保存在MessageQueue的Java文件里,前面提到过:
//NativeMessageQueue对象的地址,MessageQueue的等待和唤醒机制是在JNI下执行的 @SuppressWarnings("unused") private long mPtr; // used by native code
void NativeMessageQueue::pollOnce(JNIEnv* env, int timeoutMillis) { mInCallback = true; mLooper->pollOnce(timeoutMillis); mInCallback = false; if (mExceptionObj) { env->Throw(mExceptionObj); env->DeleteLocalRef(mExceptionObj); mExceptionObj = NULL; } }调用到了Looper的pollOnce:
/system/core/libutils/Looper.cpp
int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) { ...... 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]; //整理调用epoll_wait进行等待, eventCount = 0的话表示已经超时. //timeoutMillis 是从java层传递下来的nextPollTimeoutMillis. 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_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: ; //这里是响应C++层的Handler消息,暂且不管 ...... return result; }
epoll_wait接触等待在这里有两个操作途径,一个是将等待时间设置为0, 即java层的
nextPollTimeoutMillis = 0;
另一个是在
nextPollTimeoutMillis = -1;
的情况下,调用nativeWake,这个方法最终调用到了:
Looper.cpp中的:
void Looper::wake() { #if DEBUG_POLL_AND_WAKE ALOGD("%p ~ wake", this); #endif uint64_t inc = 1; //write方法是写入了一个wake标志唤醒epoll_wair等待, 其实就是不断写入一个无用数据直到成功,来唤醒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)); } } }
rebuildEpollLocked: 设置mWakeEventFd事件监听:
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; //这里为epoll添加了mWakeEventFd事件的监听, 监听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)); ...... }
总结:
Barrier
MessageQueue中可以设置Barrier(屏障), 来阻止Barrier设置的时间点之后的同步Message执行,直到调用removeSyncBarrier被调用:
Barrier的实质是一个tartget(Handler)为空的Message对象。
返回值是当前设置的barrier的token, 这个用来在调用removeSyncBarrier的时候作为参数来取消对应的MessageQueue中的Barrier。
int enqueueSyncBarrier(long when) { // Enqueue a new sync barrier token. // We don't need to wake the queue because the purpose of a barrier is to stall it. synchronized (this) { final int token = mNextBarrierToken++; //创建一个tartget为null(未设置tartget)的msg作为barrier final Message msg = Message.obtain(); msg.markInUse(); msg.when = when; msg.arg1 = token; Message prev = null; Message p = mMessages; if (when != 0) { //执行时间在barrier之前的message不会被阻止执行. while (p != null && p.when <= when) { prev = p; p = p.next; } } if (prev != null) { // invariant: p == prev.next //如果有将要执行的message(行时间在barrier之前),会将barrier放在这个message之后,这个message不会被阻止执行 msg.next = p; prev.next = msg; } else { //没有将要执行的message, 则直接将barrier放在头部,后面的message都不会被执行,直到调用removeSyncBarrier msg.next = p; mMessages = msg; } return token; } }
removeSyncBarrier:
void removeSyncBarrier(int token) { // Remove a sync barrier token from the queue. // If the queue is no longer stalled by a barrier then wake it. synchronized (this) { Message prev = null; Message p = mMessages; while (p != null && (p.target != null || p.arg1 != token)) { prev = p; p = p.next; } if (p == null) { throw new IllegalStateException("The specified message queue synchronization " + " barrier token has not been posted or has already been removed."); } final boolean needWake; if (prev != null) { prev.next = p.next; needWake = false; } else { mMessages = p.next; needWake = mMessages == null || mMessages.target != null; } p.recycleUnchecked(); // If the loop is quitting then it is already awake. // We can assume mPtr != 0 when mQuitting is false. if (needWake && !mQuitting) { nativeWake(mPtr); } } }