Handler详解 · Pride's Blog

Android有大量的消息驱动方式来进行交互，比如Android的四大组件Activity, Service, Broadcast, ContentProvider的启动过程的交互，都离不开消息机制，Android某种意义上也可以说成是一个以消息驱动的系统。消息机制涉及MessageQueue/Message/Looper/Handler这4个类。

一、概念§

Handler

Message: 消息分为硬件产生的消息(如按钮、触摸)和软件生成的消息，封装成为消息体，包含一个Handler对象用来处理消息；
MessageQueue: 消息队列内部构成为一个单链表，将Message链接在一起，其主要功能向消息池投递消息(MessageQueue.enqueueMessage)和取走消息池的消息(MessageQueue.next)；
Handler: Handle 消息机制中作为一个对外暴露的工具，其内部包含了一个 Looper；主要功能向消息池发送各种消息事件(Handler.sendMessage)和处理相应消息事件(Handler.handleMessage)；
Looper: Looper 作为消息循环的核心，其内部包含了一个消息队列 MessageQueue ，用于记录所有待处理的消息；不断循环执行(Looper.loop)，按分发机制将消息分发给目标处理者。

二、Looper§

1.prepare()§

prepare()方法会默认调用prepare(true)，表示该Looper允许退出，反之对于false的情况表示Looper不允许退出。每个线程只允许执行一次此方法，第二次执行时现成的TLS已有数据，就会抛出异常。

    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");
        }
        // 创建Looper对象，并保存到当前线程的TLS中
        sThreadLocal.set(new Looper(quitAllowed));
    }

sThreadLocal又是什么鬼？咱们看一下它的定义。

可以看到这个sThreadLocal是一个ThreadLocal类，并且它的泛型是Looper对象。

ThreadLocal: 线程本地存储区（Thread Local Storage，简称为TLS），每个线程都有自己的私有的本地存储区域，不同线程之间彼此不能访问对方的TLS区域。提供了线程的局部变量，每个线程都可以通过set()和get()来对这个局部变量进行操作，但不会和其他线程的局部变量进行冲突，实现了线程的数据隔离。简要言之：往ThreadLocal中填充的变量属于当前线程，该变量对其他线程而言是隔离的。

ThreadLocal.set(T value)：将value存储到当前线程的TLS区域

    public void set(T value) {
        // 获取当前线程
        Thread currentThread = Thread.currentThread(); 
        // 查找当前线程的本地储存区
        Values values = values(currentThread); 
        if (values == null) {
            // 当线程本地存储区，尚未存储该线程相关信息时，则创建Values对象
            values = initializeValues(currentThread);
        }
        // 保存数据value到当前线程this
        values.put(this, value);
    }

ThreadLocal.get()：获取当前线程TLS区域的数据

    public T get() {
        // 获取当前线程
        Thread currentThread = Thread.currentThread(); 
        // 查找当前线程的本地储存区
        Values values = values(currentThread); 
        if (values != null) {
            Object[] table = values.table;
            int index = hash & values.mask;
            if (this.reference == table[index]) {
                // 返回当前线程储存区中的数据
                return (T) table[index + 1]; 
            }
        } else {
            // 创建Values对象
            values = initializeValues(currentThread);
        }
        // 从目标线程存储区没有查询是则返回null
        return (T) values.getAfterMiss(this); 
    }

    private Looper(boolean quitAllowed) {
        // 创建MessageQueue对象
        mQueue = new MessageQueue(quitAllowed);
        // 记录当前线程
        mThread = Thread.currentThread();
    }

其中还有一个prepareMainLooper()方法，主要在ActivityThread中使用

    public static void prepareMainLooper() {
        // 设置不允许退出的Looper
        prepare(false);
        synchronized (Looper.class) {
            // 将当前的Looper保存为主Looper，每个线程只允许执行一次。
            if (sMainLooper != null) {
                throw new IllegalStateException("The main Looper has already been prepared.");
            }
            sMainLooper = myLooper();
        }
    }

Looper.prepare()会创建Looper对象，Looper的构造方法中会创建一个MessageQueue对象，再将Looper对象保存到当前线程TLS。

2.loop()§

    public static void loop() {
        // 获取TLS存储的Looper对象
        final Looper me = myLooper();
        if (me == null) {
            throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");
        }
        // 获取Looper对象中的消息队列
        final MessageQueue queue = me.mQueue;

        Binder.clearCallingIdentity();
        final long ident = Binder.clearCallingIdentity();

        final int thresholdOverride =
                SystemProperties.getInt("log.looper."
                        + Process.myUid() + "."
                        + Thread.currentThread().getName()
                        + ".slow", 0);

        boolean slowDeliveryDetected = false;

        // 进入loop的主循环方法
        for (;;) {
            Message msg = queue.next(); // 可能会阻塞
            if (msg == null) {
                // 没有消息，则退出循环
                return;
            }

            // 默认为null，可通过setMessageLogging()方法来指定输出，用于debug功能
            final Printer logging = me.mLogging;
            if (logging != null) {
                logging.println(">>>>> Dispatching to " + msg.target + " " +
                        msg.callback + ": " + msg.what);
            }

            final long traceTag = me.mTraceTag;
            long slowDispatchThresholdMs = me.mSlowDispatchThresholdMs;
            long slowDeliveryThresholdMs = me.mSlowDeliveryThresholdMs;
            if (thresholdOverride > 0) {
                slowDispatchThresholdMs = thresholdOverride;
                slowDeliveryThresholdMs = thresholdOverride;
            }
            final boolean logSlowDelivery = (slowDeliveryThresholdMs > 0) && (msg.when > 0);
            final boolean logSlowDispatch = (slowDispatchThresholdMs > 0);

            final boolean needStartTime = logSlowDelivery || logSlowDispatch;
            final boolean needEndTime = logSlowDispatch;

            if (traceTag != 0 && Trace.isTagEnabled(traceTag)) {
                Trace.traceBegin(traceTag, msg.target.getTraceName(msg));
            }

            final long dispatchStart = needStartTime ? SystemClock.uptimeMillis() : 0;
            final long dispatchEnd;
            try {
                // 通过消息处理器分发消息
                msg.target.dispatchMessage(msg);
                dispatchEnd = needEndTime ? SystemClock.uptimeMillis() : 0;
            } finally {
                if (traceTag != 0) {
                    Trace.traceEnd(traceTag);
                }
            }
            if (logSlowDelivery) {
                if (slowDeliveryDetected) {
                    if ((dispatchStart - msg.when) <= 10) {
                        Slog.w(TAG, "Drained");
                        slowDeliveryDetected = false;
                    }
                } else {
                    if (showSlowLog(slowDeliveryThresholdMs, msg.when, dispatchStart, "delivery",
                            msg)) {
                        slowDeliveryDetected = true;
                    }
                }
            }
            if (logSlowDispatch) {
                showSlowLog(slowDispatchThresholdMs, dispatchStart, dispatchEnd, "dispatch", msg);
            }

            if (logging != null) {
                logging.println("<<<<< Finished to " + msg.target + " " + msg.callback);
            }

            // 恢复调用者信息
            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放入消息池
            msg.recycleUnchecked();
        }
    }

loop()进入循环后，会不断执行以下过程，直到没有消息的时候退出循环:

读取MessageQueue的下一条Message
把Message分发给相应的消息处理器target
再把分发后的Message回收到消息池，以便重复利用

这就是这个消息处理的核心部分。

3.quit()§

    // 移除消息
    public void quit() {
        mQueue.quit(false);
    }
    
    // 安全移除消息
    public void quitSafely() {
        mQueue.quit(true);
    }

Looper.quit()方法的实现最终调用的是MessageQueue.quit()方法。

    void quit(boolean safe) {
        // 当mQuitAllowed为false
        // 表示不运行退出，强行调用quit()会抛出异常
        if (!mQuitAllowed) {
            throw new IllegalStateException("Main thread not allowed to quit.");
        }

        synchronized (this) {
            if (mQuitting) {
                // 防止多次执行退出操作
                return;
            }
            mQuitting = true;

            if (safe) {
                // 移除尚未触发的所有消息
                removeAllFutureMessagesLocked();
            } else {
                // 移除所有的消息
                removeAllMessagesLocked();
            }

            // mQuitting=false，那么认定为 mPtr != 0
            nativeWake(mPtr);
        }
    }

消息退出的方式：

当safe =true时，只移除尚未触发的所有消息，对于正在触发的消息并不移除
当safe =flase时，移除所有的消息

4.myLooper()§

    // 用于获取TLS存储的Looper对象
    public static @Nullable Looper myLooper() {
        return sThreadLocal.get();
    }

5.post()§

发送消息，并设置消息的callback，用于处理消息。

    public final boolean post(Runnable r) {
    return  sendMessageDelayed(getPostMessage(r), 0);
    }

    private static Message getPostMessage(Runnable r) {
        Message m = Message.obtain();
        m.callback = r;
        return m;
    }

三、Handler§

1.创建Handler§

a.无参构造§

对于Handler的无参构造方法，默认采用当前线程TLS中的Looper对象，并且callback回调方法为null，且消息为同步处理方式。只要执行的Looper.prepare()方法，那么便可以获取有效的Looper对象。

public Handler() {
    this(null, false);
}

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());
        }
    }
    // 必须先执行Looper.prepare()，才能获取Looper对象，否则为null.
    mLooper = Looper.myLooper();  
    // 从当前线程的TLS中获取Looper对象
    if (mLooper == null) {
        throw new RuntimeException("");
    }
    // 消息队列，来自Looper对象
    mQueue = mLooper.mQueue; 
    // 回调方法
    mCallback = callback;  
    // 设置消息是否为异步处理方式
    mAsynchronous = async; 
}

b.有参构造§

Handler类在构造方法中，可指定Looper，Callback回调方法以及消息的处理方式(同步或异步)，对于无参的handler，默认是当前线程的Looper。

public Handler(Looper looper) {
    this(looper, null, false);
}

public Handler(Looper looper, Callback callback, boolean async) {
    mLooper = looper;
    mQueue = looper.mQueue;
    mCallback = callback;
    mAsynchronous = async;
}

2.消息分发机制§

在Looper.loop()中，当发现有消息的时候，就会调用消息的目标处理器target(Handler传入对象)，执行dispatchMessage()方法来分发消息。

    /**
     * Handle system messages here.
     */
    public void dispatchMessage(Message msg) {
        if (msg.callback != null) {
            // 当Message存在回调方法，回调msg.callback.run()方法
            handleCallback(msg);
        } else {
            if (mCallback != null) {
                // 当Handler存在Callback成员变量时，回调方法handleMessage()
                if (mCallback.handleMessage(msg)) {
                    return;
                }
            }
            // Handler自身的回调方法handleMessage()
            handleMessage(msg);
        }
    }

消息分发流程:

当Message的回调方法不为空时，则回调方法msg.callback.run()，其中callback数据类型为Runnable，否则就会进入步骤2。
当Handler的mCallback成员变量不为空的时候，回调方法mCallback.handlerMessage(msg)，否则进入步骤3。
调用Handler自身的回调方法handleMessage()，该方法默认为空，Handler子类通过覆盖写入这个方法来完成具体的逻辑。

消息调用链: enqueueMessage

最终走向了一个Handler#enqueueMessage()方法

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

Handler.sendEmptyMessage()等系列方法最终调用MessageQueue.enqueueMessage(msg, uptimeMillis)，将消息添加到消息队列中，其中uptimeMillis为系统当前的运行时间，不包括休眠时间。

3.消息获取§

    public final Message obtainMessage()
    {
        return Message.obtain(this);
    }

Handler.obtainMessage()方法，最终调用Message.obtain(this)，其中this为当前的Handler对象。

4.移除消息§

    public final void removeMessages(int what) {
        mQueue.removeMessages(this, what, null); 【见 4.5】
    }

Handler是消息机制中非常重要的辅助类，更多的实现都是MessageQueue, Message中的方法，Handler的目的是为了更加方便的使用消息机制。

四、MessageQueue§

MessageQueue是消息机制的Java层和C++层的连接纽带，大部分核心方法都在这里交给native层来处理:

    private native static long nativeInit();
    private native static void nativeDestroy(long ptr);
    private native void nativePollOnce(long ptr, int timeoutMillis); 
    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);

1.native方法解析§

native

a.nativeInit()§

nativeInit native侧的初始化实现，调用过程如下:

new MessageQueue()

// MessageQueue.java

MessageQueue(boolean quitAllowed) {
    mQuitAllowed = quitAllowed;
    // mPtr记录native消息队列的信息
    mPtr = nativeInit();
}

android_os_MessageQueue_nativeInit()

// android_os_MessageQueue.cpp

static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) {
    // 初始化native消息队列
    NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue();
    // 增加引用计数
    nativeMessageQueue->incStrong(env);
    return reinterpret_cast<jlong>(nativeMessageQueue);
}

new NativeMessageQueue()

// android_os_MessageQueue.cpp

NativeMessageQueue::NativeMessageQueue()
            : mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {
    // 获取TLS中的Looper对象
    mLooper = Looper::getForThread();
    if (mLooper == NULL) {
        // 创建native层的Looper
        mLooper = new Looper(false);
        // 保存native层的Looper到TLS
        Looper::setForThread(mLooper);
    }
}

new Looper()

// Looper.cpp

Looper::Looper(bool allowNonCallbacks) :
        mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
        mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false),
        mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
    // 构造唤醒事件的fd
    mWakeEventFd = eventfd(0, EFD_NONBLOCK);
    AutoMutex _l(mLock);
    // 重建Epoll事件
    rebuildEpollLocked();
}

epoll_create/epoll_ctl

// Looper.cpp

void Looper::rebuildEpollLocked() {
    if (mEpollFd >= 0) {
        // 关闭旧的epoll实例
        close(mEpollFd);
    }
    // 创建新的epoll实例，并注册wake管道
    mEpollFd = epoll_create(EPOLL_SIZE_HINT);
    struct epoll_event eventItem;
    // 把未使用的数据区域进行置0操作
    memset(& eventItem, 0, sizeof(epoll_event));
    // 可读事件
    eventItem.events = EPOLLIN;
    eventItem.data.fd = mWakeEventFd;
    // 将唤醒事件(mWakeEventFd)添加到epoll实例(mEpollFd)
    int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);

    for (size_t i = 0; i < mRequests.size(); i++) {
        const Request& request = mRequests.valueAt(i);
        struct epoll_event eventItem;
        request.initEventItem(&eventItem);
        // 将request队列的事件，分别添加到epoll实例
        int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, & eventItem);
    }
}

Looper对象中的mWakeEventFd添加到epoll监控，以及mRequests也添加到epoll的监控范围内。

b.nativeDestroy()§

nativeDestroy 清理回收调用链流程如下:

MessageQueue.dispose()

// MessageQueue.java

private void dispose() {
    if (mPtr != 0) {
        nativeDestroy(mPtr); 【2】
        mPtr = 0;
    }
}

android_os_MessageQueue_nativeDestroy()

// android_os_MessageQueue.cpp

static void android_os_MessageQueue_nativeDestroy(JNIEnv* env, jclass clazz, jlong ptr) {
    NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
    nativeMessageQueue->decStrong(env);
}

nativeMessageQueue继承自RefBase类，所以decStrong最终调用的是RefBase.decStrong().

RefBase::decStrong()

// RefBase.cpp

void RefBase::decStrong(const void* id) const
{
    weakref_impl* const refs = mRefs;
    // 移除强引用
    refs->removeStrongRef(id);
    const int32_t c = android_atomic_dec(&refs->mStrong);
    if (c == 1) {
        refs->mBase->onLastStrongRef(id);
        if ((refs->mFlags&OBJECT_LIFETIME_MASK) == OBJECT_LIFETIME_STRONG) {
            delete this;
        }
    }
    // 移除弱引用
    refs->decWeak(id); 
}

c.nativePollOnce()§

nativePollOnce用于提取消息队列中的消息，提取消息的调用链:

MessageQueue.next()

// MessageQueue.java

Message next() {
    final long ptr = mPtr;
    if (ptr == 0) {
        return null;
    }

    for (;;) {
        // 阻塞操作
        nativePollOnce(ptr, nextPollTimeoutMillis);
    }
}

android_os_MessageQueue_nativePollOnce()

// android_os_MessageQueue.cpp

static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj, jlong ptr, jint timeoutMillis) {
    // 将Java层传递下来的mPtr转换为nativeMessageQueue
    NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
    nativeMessageQueue->pollOnce(env, obj, timeoutMillis); 【3】
}

NativeMessageQueue::pollOnce()

// android_os_MessageQueue.cpp

void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) {
    mPollEnv = env;
    mPollObj = pollObj;
    mLooper->pollOnce(timeoutMillis); 【4】
    mPollObj = NULL;
    mPollEnv = NULL;
    if (mExceptionObj) {
        env->Throw(mExceptionObj);
        env->DeleteLocalRef(mExceptionObj);
        mExceptionObj = NULL;
    }
}

Looper::pollOnce()

// Looper.h

inline int pollOnce(int timeoutMillis) {
    return pollOnce(timeoutMillis, NULL, NULL, NULL);
}

Looper::pollOnce()

// Looper.cpp
/*
 * timeoutMillis：超时时长
 * outFd：发生事件的文件描述符
 * outEvents：当前outFd上发生的事件，包含以下4类事件
 *    EVENT_INPUT 可读
 *    EVENT_OUTPUT 可写
 *    EVENT_ERROR 错误
 *    EVENT_HANGUP 中断
 * outData：上下文数据
 * 
*/

int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {
    int result = 0;
    for (;;) {
        // 先处理没有Callback方法的 Response事件
        while (mResponseIndex < mResponses.size()) {
            const Response& response = mResponses.itemAt(mResponseIndex++);
            int ident = response.request.ident;
            if (ident >= 0) { 
                // ident大于0，则表示没有callback, 因为POLL_CALLBACK = -2,
                int fd = response.request.fd;
                int events = response.events;
                void* data = response.request.data;
                if (outFd != NULL) *outFd = fd;
                if (outEvents != NULL) *outEvents = events;
                if (outData != NULL) *outData = data;
                return ident;
            }
        }
        if (result != 0) {
            if (outFd != NULL) *outFd = 0;
            if (outEvents != NULL) *outEvents = 0;
            if (outData != NULL) *outData = NULL;
            return result;
        }
        // 再处理内部轮询
        result = pollInner(timeoutMillis); 【6】
    }
}

Looper::pollInner()

// Looper.cpp
/*
 * POLL_WAKE： 表示由wake()触发，即pipe写端的write事件触发；
 * POLL_CALLBACK： 表示某个被监听fd被触发。
 * POLL_TIMEOUT： 表示等待超时；
 * POLL_ERROR：表示等待期间发生错误；
 */
int Looper::pollInner(int timeoutMillis) {
    ...
    int result = POLL_WAKE;
    mResponses.clear();
    mResponseIndex = 0;
    // 即将处于idle状态
    mPolling = true; 
    // fd最大个数为16
    struct epoll_event eventItems[EPOLL_MAX_EVENTS]; 
    // 等待事件发生或者超时，在nativeWake()方法，向管道写端写入字符，则该方法会返回；
    int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
    // 不再处于idle状态
    mPolling = false; 
    mLock.lock();  
    // 请求锁
    if (mEpollRebuildRequired) {
        mEpollRebuildRequired = false;
        rebuildEpollLocked();  
        // epoll重建，直接跳转Done;
        goto Done;
    }
    if (eventCount < 0) {
        if (errno == EINTR) {
            goto Done;
        }
        result = POLL_ERROR; 
        // epoll事件个数小于0，发生错误，直接跳转Done;
        goto Done;
    }
    if (eventCount == 0) {  
        // epoll事件个数等于0，发生超时，直接跳转Done;
        result = POLL_TIMEOUT;
        goto Done;
    }

    // 循环遍历，处理所有的事件
    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 {
            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;
                // 处理request，生成对应的reponse对象，push到响应数组
                pushResponse(events, mRequests.valueAt(requestIndex));
            }
        }
    }
Done: ;
    // 再处理Native的Message，调用相应回调方法
    mNextMessageUptime = LLONG_MAX;
    while (mMessageEnvelopes.size() != 0) {
        nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
        const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0);
        if (messageEnvelope.uptime <= now) {
            {
                sp<MessageHandler> handler = messageEnvelope.handler;
                Message message = messageEnvelope.message;
                mMessageEnvelopes.removeAt(0);
                mSendingMessage = true;
                // 释放锁
                mLock.unlock(); 
                // 处理消息事件
                handler->handleMessage(message);  
            }
            // 请求锁
            mLock.lock();  
            mSendingMessage = false;
            // 发生回调
            result = POLL_CALLBACK; 
        } else {
            mNextMessageUptime = messageEnvelope.uptime;
            break;
        }
    }
    // 释放锁
    mLock.unlock(); 

    //处理带有Callback()方法的Response事件，执行Reponse相应的回调方法
    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;
            // 处理请求的回调方法
            int callbackResult = response.request.callback->handleEvent(fd, events, data);
            if (callbackResult == 0) {
                removeFd(fd, response.request.seq); //移除fd
            }
            response.request.callback.clear(); //清除reponse引用的回调方法
            result = POLL_CALLBACK;  // 发生回调
        }
    }
    return result;
}

Looper::awoken()

void Looper::awoken() {
    uint64_t counter;
    //不断读取管道数据，目的就是为了清空管道内容
    TEMP_FAILURE_RETRY(read(mWakeEventFd, &counter, sizeof(uint64_t)));
}

d.nativeWake()§

nativeWake用于唤醒功能，在添加消息到消息队列enqueueMessage(), 或者把消息从消息队列中全部移除quit()，再有需要时都会调用 nativeWake方法。

MessageQueue.enqueueMessage()

// MessageQueue.java

boolean enqueueMessage(Message msg, long when) {
    // 将Message按时间顺序插入MessageQueue
    if (needWake) {
            nativeWake(mPtr);
        }
}

android_os_MessageQueue_nativeWake()

// 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()

// android_os_MessageQueue.cpp

void NativeMessageQueue::wake() {
    mLooper->wake();
}

Looper::wake()

// Looper.cpp

void Looper::wake() {
    uint64_t inc = 1;
    // 向管道mWakeEventFd写入字符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);
        }
    }
}

这里会调用一段宏定义如下

/*
 * The GNU library provides a convenient way to retry a call after a temporary failure, with the macro TEMP_FAILURE_RETRY: — Macro: TEMP_FAILURE_RETRY (expression)
 * This macro evaluates expression once, and examines its value as type long int. If the value equals -1, that indicates a failure and errno should be set to show what kind of failure. If it fails and reports error code EINTR, TEMP_FAILURE_RETRY evaluates it again, and over and over until the result is not a temporary failure. The value returned by TEMP_FAILURE_RETRY is whatever value expression produced. 
*/
#define TEMP_FAILURE_RETRY(expression) \  
  (__extension__\  
   ({ long int __result;\  
       do __result = (long int)(expression);\  
       while(__result == -1L&& errno == EINTR);\  
       __result;})\  
#endif

这段宏定义在执行write操作的时候，如果执行失败，它会不断重复执行直至执行成功为止。

e.sendMessage§

Java层向MessageQueue类中添加消息后会走向Native层，调用Native层的方法进行处理，下面会一步步讲解Native是如何处理Java层传过来的消息的。

sendMessage

void Looper::sendMessage(const sp<MessageHandler>& handler, const Message& message) {
    // 取当前系统时间
    nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
    // 将当前时间和需要处理的消息以及消息处理器一并下发
    sendMessageAtTime(now, handler, message);
}

sendMessageDelayed

void Looper::sendMessageDelayed(nsecs_t uptimeDelay, const sp<MessageHandler>& handler,
        const Message& message) {
    // 取当前系统时间
    nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
    // 将当前时间加上要延迟的毫秒和需要处理的消息以及消息处理器一并下发
    sendMessageAtTime(now + uptimeDelay, handler, message);
}

-sendMessageAtTime 最后都走向了sendMessageAtTime这个方法来完成消息的插入

void Looper::sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler>& handler,
        const Message& message) {
    size_t i = 0;
    { 
        // 请求锁
        AutoMutex _l(mLock);
        size_t messageCount = mMessageEnvelopes.size();
        // 通过比对时间找到message应该插入的位置i
        while (i < messageCount && uptime >= mMessageEnvelopes.itemAt(i).uptime) {
            i += 1;
        }
        MessageEnvelope messageEnvelope(uptime, handler, message);
        mMessageEnvelopes.insertAt(messageEnvelope, i, 1);
        // 如果当前正在发送消息，那么不再调用wake()，直接返回
        if (mSendingMessage) {
            return;
        }
    } // 释放锁
    // 当把消息加入到消息队列的头部时，需要唤醒poll循环
    if (i == 0) {
        wake();
    }
}

流程总结:§

nativeInit():
- 创建了NativeMessage对象，增加引用计数，将NativeMessageQueue的指针mPtr保存在Java层的MessageQueue
- 创建了Native Looper对象
- 调用了epoll的epoll_create()/epoll_ctl()来完成对mWakeEventFd和mRequests的可读事件监听
nativeDestroy():
- 调用了RefBase::decStrong()来减少对象的引用计数
- 当引用计数为0时，就会清除掉NativeMessageQueue对象
nativePollOnce():
- 调用Looper::pollOnce()来完成，空闲时停留在epoll_wait()方法，用于等待事件发生或者超时
nativeWake()
- 调用Looper::wake()来完成，向管道mWakeEventfd写入字符，来更新状态

2.创建MessageQueue§

    MessageQueue(boolean quitAllowed) {
        mQuitAllowed = quitAllowed;
        // 通过native方法初始化消息队列，其中mPtr是供native代码使用
        mPtr = nativeInit();
    }

3.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.
                        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.
                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;
                }

                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 {
                    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;
        }
    }

nativePollOnce是阻塞操作，其中nextPollTimeoutMillis代表下一个消息到来前，还需要等待的时长；当nextPollTimeoutMillis = -1时，表示消息队列中无消息，会一直等待下去。

当处于空闲时，往往会执行IdleHandler中的方法。当nativePollOnce()返回后，next()从mMessages中提取一个消息。

IdleHandler接口被定义在 MessageQueue 中.

    public static interface IdleHandler {
        /**
         * Called when the message queue has run out of messages and will now
         * wait for more.  Return true to keep your idle handler active, false
         * to have it removed.  This may be called if there are still messages
         * pending in the queue, but they are all scheduled to be dispatched
         * after the current time.
         */
        boolean queueIdle();
    }

定义时需要实现其 queueIdle() 方法。同时返回值为 true 表示是一个持久的 IdleHandler 会重复使用，返回 false 表示是一个一次性的 IdleHandler。

同时也定义为相应的add和remove方法

    public void addIdleHandler(@NonNull IdleHandler handler) {
        if (handler == null) {
            throw new NullPointerException("Can't add a null IdleHandler");
        }
        synchronized (this) {
            mIdleHandlers.add(handler);
        }
    }

    public void removeIdleHandler(@NonNull IdleHandler handler) {
        synchronized (this) {
            mIdleHandlers.remove(handler);
        }
    }

可以看到add和remove其实都是在操作mIdleHandlers，其数据类型是一个ArrayList。

    private final ArrayList<IdleHandler> mIdleHandlers = new ArrayList<IdleHandler>();

IdleHandler 主要是在 MessageQueue 出现空闲的时候被执行，MessageQueue 是一个基于消息触发时间的优先级队列，所以队列出现空闲存在两种场景:

MessageQueue为空，没有消息。
MessageQueue中最近需要处理的消息，是一个延迟消息，需要滞后执行。

4.enqueueMessage§

    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;
    }

MessageQueue是按照Message触发时间的先后顺序排列，队列头部的消息是将会被最先触发的消息，当有消息需要加入消息队列的时候，会从队列头部开始遍历，直到找到合适的位置才会插入，保证所有消息的时间顺序。

5.removeMessages§

    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;
            }
        }
    }

    void removeMessages(Handler h, Runnable r, Object object) {
        if (h == null || r == null) {
            return;
        }

        synchronized (this) {
            Message p = mMessages;

            // Remove all messages at front.
            while (p != null && p.target == h && p.callback == r
                   && (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.callback == r
                        && (object == null || n.obj == object)) {
                        Message nn = n.next;
                        n.recycleUnchecked();
                        p.next = nn;
                        continue;
                    }
                }
                p = n;
            }
        }
    }

这个移除消息的方法，采用了两个while循环，第一个循环是从队头开始，移除符合条件的消息，第二个循环是从头部移除完连续的满足条件的消息之后，再从队列后面继续查询是否有满足条件的消息需要被移除。

6.postSyncBarrier§

    public int postSyncBarrier() {
        return postSyncBarrier(SystemClock.uptimeMillis());
    }

    private int postSyncBarrier(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++;
            final Message msg = Message.obtain();
            msg.markInUse();
            msg.when = when;
            msg.arg1 = token;

            Message prev = null;
            Message p = mMessages;
            if (when != 0) {
                while (p != null && p.when <= when) {
                    prev = p;
                    p = p.next;
                }
            }
            if (prev != null) { // invariant: p == prev.next
                msg.next = p;
                prev.next = msg;
            } else {
                msg.next = p;
                mMessages = msg;
            }
            return token;
        }
    }

每一个普通Message必须有一个target，对于特殊的message是没有target，即同步barrier token。这个消息的价值就是用于拦截同步消息，所以并不会唤醒Looper。

    public 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);
            }
        }
    }

postSyncBarrier只对同步消息产生影响，对于异步消息没有任何差别。

五、Message§

1.消息构成§

变量名	类型	含义
what	int	类别
when	long	触发时间
arg1 / arg2	int	参数
obj	Object	消息内容
target	Handler	消息处理器
callback	Runnable	回调方法

2.消息池§

静态变量sPool的数据类型为Message，通过next成员变量，维护一个消息池；静态变量MAX_POOL_SIZE是指消息池的可用大小，默认为50，消息池常用方法是obtain()和recycle()，有什么用？这在我另一篇文章中有描述。

a.obtain§

    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;
            }
        }
        return new Message();
    }

b.recycle§

    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.
     */
    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++;
            }
        }
    }