Android消息机制探索(Handler,Looper,Message,MessageQueue)

 概览

        Android消息机制是Android操作系统中比较重要的一块。具体使用方法在这里不再阐述，可以参考Android的官方开发文档。

消息机制的主要用途有两方面：

        1、线程之间的通信。比如在子线程中想更新UI，就通过发送更新消息到UI线程中来实现。

        2、任务延迟执行。比如30秒后执行刷新任务等。

消息机制运行的大概示意图如下：

        一个线程中只能有一个Looper对象，一个Looper对象中持有一个消息队列，一个消息队列中维护多个消息对象，用一个Looper可以创建多个Handler对象，Handler对象用来发送消息到消息队列中、和处理Looper分发给自己的消息（也就是自己之前发送的那些消息），这些Handler对象可以跨线程运行，但是最终消息的处理，都是在创建该Handler的线程中运行。 

 

分析

        在熟悉了基本用法之后，有必要深入探索一下。

逻辑分析

        Android消息机制的framework层主要围绕Handler、Looper、Message、MessageQueue这四个对象来操作。消息机制主要是对消息进行生成、发送、存储、分发、处理等操作。

Message：

        该类代表的是消息机制中的消息对象。是在消息机制中被创建，用来传递数据以及操作的对象，也负责维护消息对象缓存池。

Message对象中主要有以下几个属性：

what：消息类型。

arg1、arg2、obj、data：该消息的数据域

when：该消息应该被处理的时间，该字段的值为SystemClock.uptimeMillis()。当该字段值为0时，说明该消息需要被放置到消息队列的首部。

target：发送和处理该消息的Handler对象。

next：对象池中该消息的下一个消息。

        Message对象中，主要维护了一个Message对象池，因为系统中会频繁的使用到Message对象，所以用对象池的方式来减少频繁创建对象带来的开支。Message对象池使用单链表实现。最大数量限制为50。所以官方推荐我们通过对象池来获取Message对象。

        特别注意的是，我们平常使用的都是普通的Message对象，也就是同步的Message对象。其实还有两种特殊的Message对象，目前很少被使用到，但也有必要了解一下。

        第一个是同步的障碍消息（Barrier Message），该消息的作用就是，如果该消息到达消息队列的首部，则消息队列中其他的同步消息就会被阻塞，不能被处理。障碍消息的特征是target==null&&arg1==barrierToken

        第二个是异步消息，异步消息不会被上面所说的障碍消息影响。通过Message对象的setAsynchronous(boolean async)方法来设置一个消息为异步消息。

MessageQueue：

        该类代表的是消息机制中的消息队列。它主要就是维护一个线程安全的消息队列，提供消息的入队、删除、以及阻塞方式的轮询取出等操作。

Looper：

         该类代表的是消息机制中的消息分发器。 有了消息，有了消息队列，还缺少处理消息分发机制的对象，Looper就是处理消息分发机制的对象。它会把每个Message发送到正确的处理对象上进行处理。如果一个Looper开始工作后，一直没有消息处理的话，那么该线程就会被阻塞。在非UI线程中，这时候应该监听当前MessageQueue的Idle事件，如果当前有Idle事件，则应该退出当前的消息循环，然后结束该线程，释放相应的资源。

Handler：

        该类代表的是消息机制中的消息发送和处理器。有了消息、消息队列、消息分发机制，还缺少的就是消息投递和消息处理。Handler就是用来做消息投递和消息处理的。Handler事件处理机制采用一种按自由度从高到低的优先级进行消息的处理，正常情况下，一个Handler对象可以设置一个callback属性，一个Handler对象可以操作多个Message对象，从某种程度上来说，创建一个Message对象比给一个Handler对象设置callback属性来的自由，而给一个Handler对象设置callback属性比衍生一个Handler子类来的自由，所以消息处理优先级为Message>Handler.callback.Handler.handleMessage()。

代码分析

        重要部分的源代码解析，源代码基于sdk 23。

Message：

        普通的消息对象，包含了消息类型、数据、行为。内部包含了一个用单链表实现的对象池，最大数量为50，为了避免频繁的创建对象带来的开销。

1、从对象池中获取Message对象。

源代码:

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

伪代码:

public static Message obtain() {
        synchronized (sPoolSync) {
            if (对象池不为空) {
                从单链表实现的对象池中取出一个对象（从链表头获取）;
        清空该对象标志位（在使用中、异步等）;
                修正对象池大小;
                return 取出的消息对象;
            }
        }
        return 新建消息对象;
    }

 

2、返回Message对象到对象池

源代码: 

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

伪代码:

void recycleUnchecked() {
        标志为正在使用中;
        清空当前对象的其他数据;

        synchronized (sPoolSync) {
            if (对象池没容量有达到上限) {
                在单链表表头插入该对象;
                修正对象池大小;
            }
        }
    }

Looper：

        使用ThreadLocal来实现线程作用域的控制，每个线程最多有一个Looper对象，内部持有一个MessageQueue的引用。

1、初始化一个Looper

源代码:

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

伪代码:

    private static void prepare(boolean quitAllowed) {
        if (当前线程中已经有了一个Looper对象) {
            throw new RuntimeException("一个线程只能创建一个Looper对象");
        }
        重新实例化一个可以退出的Looper;
        把该Looper对象和当前线程关联起来;
    }

 

2、Looper开始工作

源代码:

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

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

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

伪代码:

   public static void loop() {
        获取当前线程的Looper对象;
        if (当前线程没有Looper对象) {
            throw new RuntimeException("没有Looper; Looper.prepare() 没有在当前线程被调用过");
        }
        获取该Looper对象关联的MessageQueue;

        清除IPC身份标志;

        for (;;) {
            从MessageQueue中获取一个Message，如果当前MessageQueue没有消息，就会阻塞;
            if (没有取到消息) {
                // 没有消息意味着消息队列退出了.
                return;
            }

            打印日志;
            
            调用当前Message对象的target来处理消息，也就是发送该Message的Handler对象;

            打印日志;

            获取新的IPC身份标识;
            if (IPC身份标识改变了) {
                打印警告信息;

            回收该消息，放入到Message对象池中;
        }
    }

 

Handler：

        负责消息的发送、定时发送、延迟发送、消息处理等动作。

1、事件处理

源代码:

public void dispatchMessage(Message msg) {
        if (msg.callback != null) {
            handleCallback(msg);
        } else {
            if (mCallback != null) {
                if (mCallback.handleMessage(msg)) {
                    return;
                }
            }
            handleMessage(msg);
        }
    }

伪代码:

public void dispatchMessage(Message msg) {
        if (该消息的callback属性不为空) {
            运行该消息的callback对象的run()方法;
        } else {
            if (当前Handler对象的mCallback属性不为空) {
                if (mCallback对象成功处理了消息) {
                    return;
                }
            }
            Handler内部处理该消息;
        }
    }

MessageQueue：

        使用单链表的方式维护一个消息队列，提高频繁插入删除消息等操作的性能，该链表用消息的when字段进行排序，先被处理的消息排在链表前部。内部的阻塞轮询和唤醒等操作，使用JNI来实现。

1、Message对象的入队操作

源代码:

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

伪代码:

boolean enqueueMessage(Message msg, long when) {
        if (该消息没有target) {
            throw new IllegalArgumentException("Message对象必须要有一个target");
        }
        if (该消息正在被使用) {
            throw new IllegalStateException(msg + " 该消息正在使用中");
        }

        synchronized (this) {
            if (消息队列退出了) {
                打印警告信息;
                回收该消息，返回到Message对象池;
                return false;
            }

            设置该消息为正在使用中;
            设置消息将要被处理的时间;
            Message p = mMessages;
            boolean needWake;
            if (msg队列为空 || 该消息对象请求放到队首 || 执行时间先于当前队首msg的执行时间（当前队列中全是delay msg）) {
                把当前msg添加到msg队列首部;
                如果阻塞了，设置为需要被唤醒;
            } else {
                if (阻塞了 && 队首是barrier && 当前msg是异步msg) {
                    设置为需要被唤醒
                }
                for (;;) {
                    根据msg.when的先后，找到合适的插入位置，先执行的在队列前面;
                    if (需要唤醒 && 插入位置之前有异步消息) {
                        不需要唤醒;
                    }
                }
                插入到合适的位置;
            }

            if (需要唤醒) {
                调用native方法进行本地唤醒;
            }
        }
        return true;
    }

2、查询待处理消息

源代码:

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

伪代码:

Message next() {
        if (消息队列退出了) {
            return null;
        }

        把Idle事件的次数标记为第一次;
        下一次轮询的等待（阻塞）时间设为0;
        for (;;) {
            if (下一次轮询需要阻塞) {
                清楚Binder的pending command，用来释放资源;
            }

            使用当前的设置轮询阻塞时间去做一次native的轮询，如果阻塞时间大于0，则会阻塞，直到取到消息为止;

            synchronized (this) {
                if (消息队列首部为barrier消息) {
                    取出第一个异步消息;
                }
                if (查询到满足条件的消息) {
                    if (还没到该消息的执行时间) {
                        设置下一次轮询的阻塞时间为msg.when - now，最大不超过Integer.MAX_VALUE;
                    } else {
                        阻塞标识设置为false;
                        取出该消息，重定向链表头;
                        标记该消息为在使用中;
                        return 该消息;
                    }
                } else {
                    没有消息，设置下一次轮询阻塞时间为-1，不阻塞;
                }

                if (消息队列退出了) {
                    释放资源;
                    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 (第一次Idle事件) {
                    计算Idle监听器数量;
                }
                if (没有Idle监听器) {
                    阻塞标识设置为true;
                    continue;
                }

                生成Idle监听对象;
            }

            for (int i = 0; i < pendingIdleHandlerCount; i++) {
                通知Idle事件的监听对象，根据标识来确定这些监听器是否继续监听。
            }

            设置Idle事件的标识为不是第一次;

            调用了Idle监听器之后，可能有新的消息进入队列，所以下一次轮询阻塞时间设置为0;
        }
    }