Android线程的消息队列

一、子线程创建Handler的前题条件是什么

　　子线程创建Handler必须先创建子线程Looper。

　　假如，在子线程不先创建Looper会怎样？

Thread {
    _handler = Handler()
}.start()

　　结果：

E/AndroidRuntime: FATAL EXCEPTION: Thread-7
    Process: com.example.demoapp, PID: 8456
    java.lang.RuntimeException: Can't create handler inside thread Thread[Thread-7,5,main] that has not called Looper.prepare()
        at android.os.Handler.<init>(Handler.java:205)
        at android.os.Handler.<init>(Handler.java:118)
        at com.example.demoapp.MainActivity2.onCreate$lambda-0(MainActivity2.kt:17)
        at com.example.demoapp.MainActivity2.lambda$Qqlg1xygUvXBrkBzlVLkUUn-c9c(Unknown Source:0)
        at com.example.demoapp.-$$Lambda$MainActivity2$Qqlg1xygUvXBrkBzlVLkUUn-c9c.run(Unknown Source:2)
        at java.lang.Thread.run(Thread.java:764)

 　　原因，从Handler构造函数源码：

public Handler(@Nullable Callback callback, boolean async) {
    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 " + Thread.currentThread()
                    + " that has not called Looper.prepare()");
    }
    mQueue = mLooper.mQueue;
    mCallback = callback;
    mAsynchronous = async;
}

　　创建Handler的线程没有创建Looper，调用Looper.prepare()就是抛出异常。Looper是Android提供线程间通信Handler最重要的组成部分，Handler发送的消息都需要Looper进行处理。

二、主线程Looper与子线程Looper区别

 　　主线程启动Looper调用Looper.prepareMainLooper()函数：

public static void main(String[] args) {
    ……
    Looper.prepareMainLooper();
    ……
}

　　Looper.prepareMainLooper()函数实现：

/**
 * Initialize the current thread as a looper, marking it as an
 * application's main looper. See also: {@link #prepare()}
 *
 * @deprecated The main looper for your application is created by the Android environment,
 *   so you should never need to call this function yourself.
 */
@Deprecated
public static void prepareMainLooper() {
    prepare(false);
    synchronized (Looper.class) {
        if (sMainLooper != null) {
            throw new IllegalStateException("The main Looper has already been prepared.");
        }
        sMainLooper = myLooper();
    }
}

　　子线程创建Looper调用prepare()函数：

/** Initialize the current thread as a looper.
 * This gives you a chance to create handlers that then reference
 * this looper, before actually starting the loop. Be sure to call
 * {@link #loop()} after calling this method, and end it by calling
 * {@link #quit()}.
*/
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));
}

　　1. 主线程和子线程最终调用函数prepare(boolean quitAllowed)，主线程调用prepare(false)，Looper不允许退出，子线程调用prepare(true)，Looper允许退出。

　　2. 主线程Looper除了存储在sThreadLocal中，还赋值给了静态变量sMainLooper，通过getMainLooper()函数返回的就是这个静态变量。

/**
 * Returns the application's main looper, which lives in the main thread of the application.
 */
public static Looper getMainLooper() {
    synchronized (Looper.class) {
        return sMainLooper;
    }
}

三、消息队列是怎么创建的

　　在调用prepare(boolean quitAllowed)函数时，创建Looper对象实例，在Looper构造函数中创建消息队列MessageQueue。

private Looper(boolean quitAllowed) {
    mQueue = new MessageQueue(quitAllowed);
    mThread = Thread.currentThread();
}

四、Looper与消息队列的关系

　　

五、Hnadler延迟消息实现原理

　　Handler中发送延迟消息函数sendMessageDelayed()：

/**
 * Enqueue a message into the message queue after all pending messages
 * before (current time + delayMillis). You will receive it in
 * {@link #handleMessage}, in the thread attached to this handler.
 *  
 * @return Returns true if the message was successfully placed in to the 
 *         message queue.  Returns false on failure, usually because the
 *         looper processing the message queue is exiting.  Note that a
 *         result of true does not mean the message will be processed -- if
 *         the looper is quit before the delivery time of the message
 *         occurs then the message will be dropped.
 */
public final boolean sendMessageDelayed(@NonNull Message msg, long delayMillis) {
    if (delayMillis < 0) {
        delayMillis = 0;
    }
    return sendMessageAtTime(msg, SystemClock.uptimeMillis() + delayMillis);
}

/**
 * Enqueue a message into the message queue after all pending messages
 * before the absolute time (in milliseconds) <var>uptimeMillis</var>.
 * <b>The time-base is {@link android.os.SystemClock#uptimeMillis}.</b>
 * Time spent in deep sleep will add an additional delay to execution.
 * You will receive it in {@link #handleMessage}, in the thread attached
 * to this handler.
 * 
 * @param uptimeMillis The absolute time at which the message should be
 *         delivered, using the
 *         {@link android.os.SystemClock#uptimeMillis} time-base.
 *         
 * @return Returns true if the message was successfully placed in to the 
 *         message queue.  Returns false on failure, usually because the
 *         looper processing the message queue is exiting.  Note that a
 *         result of true does not mean the message will be processed -- if
 *         the looper is quit before the delivery time of the message
 *         occurs then the message will be dropped.
 */
public boolean sendMessageAtTime(@NonNull 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);
}

　　sendMessageDelayed()函数中调用sendMessageAtTime(, uptimeMillis)函数，参数uptimeMillis是消息添加的时间，消息队列是一个单向链表，按消息发送时间（ms）排序。发送延迟消息是当前时间+延迟时间，最后调用qnqueueMessage(queue, msg, uptimeMillis)函数向队列中添加消息。

private boolean enqueueMessage(@NonNull MessageQueue queue, @NonNull Message msg,
        long uptimeMillis) {
    msg.target = this;
    msg.workSourceUid = ThreadLocalWorkSource.getUid();

    if (mAsynchronous) {
        msg.setAsynchronous(true);
    }
    return queue.enqueueMessage(msg, uptimeMillis);
}

　　在Handler.enqueueMessage()函数中调用到MessageQueue.enqueueMessage()函数：

boolean enqueueMessage(Message msg, long when) {
    if (msg.target == null) {
        throw new IllegalArgumentException("Message must have a target.");
    }

    synchronized (this) {
        if (msg.isInUse()) {
            throw new IllegalStateException(msg + " This message is already in use.");
        }

        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;
        // (1)
        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;
        }

        // (2)
        // We can assume mPtr != 0 because mQuitting is false.
        if (needWake) {
            nativeWake(mPtr);
        }
    }
    return true;
}

　　在enqueueMessage()函数中标识(1)的部分逻辑，when是消息添加时间，新消息添加时间是否是0或者添加小于当前队列头部消息时间，将新消息添加到队列头，当前新消息时间大于队列头消息，在队列中找到一个适合的位置添加消息。

　　在Handler中发送延迟消息，消息是延迟处理，而非延迟添加。

六、线程间消息传递机制

　　Looper.loop是消息循环重要的函数：

/**
 * Run the message queue in this thread. Be sure to call
 * {@link #quit()} to end the loop.
 */
public static void loop() {
    final Looper me = myLooper();
    me.mInLoop = true;
    final MessageQueue queue = me.mQueue;
    ……
    for (;;) {
        // (1)
        Message msg = queue.next(); // might block
        if (msg == null) {
            // No message indicates that the message queue is quitting.
            return;
        }
        ……
        try {
            // (2)
            msg.target.dispatchMessage(msg);
            if (observer != null) {
                observer.messageDispatched(token, msg);
            }
            dispatchEnd = needEndTime ? SystemClock.uptimeMillis() : 0;
        } 
        ……
        msg.recycleUnchecked();
    }
}

　　在Looper.loop()函数中做了两件事：

• • 从队列中获取将要处理的消息。
  • 消息分发。

　　获取消息：

Message msg = queue.next(); // might block

　　　　通过queue.next()函数获取要处理的消息。

@UnsupportedAppUsage
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();
        }
        // (1)
        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;
            // (2)
            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;
    }
}

　　　　(1) nativePollOnce()函数，堵塞队列，等待有新消息加入队列，或者堵塞时间超时，检查队列是否有消息。nativePollOnce()函数是Native层函数：

　　　　　　JNI层函数，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);
}

　　　　Native层函数，NaitveMessageQueue.pollOnce()函数：

　　　　源码：frameworks/base/core/jni/android_os_MessageQueue.cpp

class NativeMessageQueue : public MessageQueue, public LooperCallback {

}

class MessageQueue : public virtual RefBase {
    public:
        /* Gets the message queue's looper. */
        inline sp<Looper> getLooper() const {
            return mLooper;
        }
        ……
    protected:
        MessageQueue();
        virtual ~MessageQueue();

    protected:
        sp<Looper> mLooper;
};

　　　　　NativeMessageQueue继承自Native层的MessageQueue，Native层的MessageQueue中mLooper变量。

　　　　(2) 在消息队列中获取消息。

　　消息分发：

　　分发消息通过msg.target.dispatchMessage(msg)函数，msg.target是发送消息的Handler对象。在Handler.dispatchMessage()函数：

/**
 * Handle system messages here.
 */
public void dispatchMessage(@NonNull Message msg) {
    // (1)
    if (msg.callback != null) {
        handleCallback(msg);
    } else {
        // (2)
        if (mCallback != null) {
            if (mCallback.handleMessage(msg)) {
                return;
            }
        }
        // (3)
        handleMessage(msg);
    }
}

　　(1) msg.callback != null，通过handleCallback()函数执行消息，msg.callback是Handler.post(runnable)的runnable函数体。Handler.post()函数将Runnable包装成一个Message，callback就是Runnable。

　　Handler.post()函数实现：

/**
 * Causes the Runnable r to be added to the message queue.
 * The runnable will be run on the thread to which this handler is 
 * attached. 
 *  
 * @param r The Runnable that will be executed.
 * 
 * @return Returns true if the Runnable was successfully placed in to the 
 *         message queue.  Returns false on failure, usually because the
 *         looper processing the message queue is exiting.
 */
public final boolean post(@NonNull Runnable r) {
    return  sendMessageDelayed(getPostMessage(r), 0);
}

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

　　m.callback = r，给runnable callback包装成Message结构。

　　(2) mCallback != null，mCallback是Handler全局的Callback，在实例化Handler时传入的Callback。通过mCallback.handleMessage(msg)，在Callback.handleMessage() == false，继续执行Handler类中重写的Handler.handleMessage()函数，这个函数前并消息清楚Message可能被修改了。通常Hook Handler的消息处理就是在这里，将Callback.handleMessage()的返回值设置false。当前返回值为ture处理消息后return。

　　(3) Handler.handleMessage()函数