Android的异步消息处理机制

异步消息处理线程的一般思路

要实现一个异步消息处理线程需要解决如下问题：

• 每个线程应该有一个消息队列，用于对消息进行排队
• 线程执行体中有一个无限的循环，不断地从消息队列中取出消息，并根据消息的来源，去调用相应的处理方法
• 其他线程可以给队列添加消息

Android通过四个主要类来实现：

• Message 封装执行的方法或携带要处理的消息参数
• MessageQueue 处理消息的排队
• Looper 不断地从MessageQueue中取出消息派发给相应的处理器
• Handler 通过它给MessageQueue发送Message，在其中执行相应的处理方法

Looper

一个Looper中持有一个MessageQueue对象，而一个线程只有一个Looper，这是怎么做到的呢？

先看Looper的构造方法：

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

我们并不能自己创建Looper对象，而是通过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));
}

原来是通过线程局部变量来实现的，保证了一个线程只能有一个Looper。当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;

    for (;;) {
        Message msg = queue.next(); // might block
        if (msg == null) {
            // No message indicates that the message queue is quitting.
            return;
        }

        msg.target.dispatchMessage(msg);

        msg.recycleUnchecked();
    }
}

可以看到，整个过程其实很简单，调用MessageQueue的next方法，取出消息队列中的一个消息，然后调用其target的dispatchMessageMessage方法，最后回收消息。

Message

Message是一个携带信息的对象，一个Message可以携带以下东西：

• int what 一般是用来表明该Message用处的标识
• int arg1和int arg2 两个简单的int值
• Object obj 一个对象
• Bundle data 一个Bundle对象，通过setData方法设置

对于Message内部运行，有如下成员变量：

/*package*/ int flags; // Message的状态

/*package*/ long when; // Message的执行时间

/*package*/ Handler target; // 处理该消息的Handler

/*package*/ Runnable callback; // 携带一个Runnable对象

// sometimes we store linked lists of these things
/*package*/ Message next; // 下一个Message

Message提供了一个public的构造方法，但并不建议我们直接使用，而是通过各种obtain方法来获取，因为Messge类本身维护了一个对象池，避免重复创建Message对象，它是怎么做到的呢？

private static final Object sPoolSync = new Object();
private static Message sPool;
private static int sPoolSize = 0;

private static final int MAX_POOL_SIZE = 50;

/**
 * Return a new Message instance from the global pool. Allows us to
 * avoid allocating new objects in many cases.
 */
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();
}

可以看到sPool就是这个池的头指针，每次从Message链表中取出一个Message返回，然后指向下一个Message。而这个Message链表是在recycleUnckecked方法中构建出来的：

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

回收完的Message插入到链表头部，设计得太巧妙了！！！

Handler

对于一个Handler，通常我们有三种用法：

• 使用sendXxx去发送一个Message
• 重写handleMessage或者设置Callback来处理发送给Handler的Message
• 使用postXxx去异步执行一个Runnable

从上图可以看到，其实各种postXxx和sendXxx最终都会调用到Handler的enqueueMessage方法。比如postXxx会把Runnable赋值给Message的callback：

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

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

而Handler的enqueueMessage最终调用MessageQueue的enqueueMessage：

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

还记得Looper中的msg.target.dispatchMessage(msg);吗？Message中的target就是与之关联的Handler，dispatchMessage的实现如下：

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

private static void handleCallback(Message message) {
    message.callback.run();
}

如果这个Message带的是一个Runnable，就直接调用run方法了，否则交给Callback或自身的handlerMessage去处理。

MessageQueue

MessageQueue的重要方法：

• next 取出队列中的下一个消息
• enqueueMessage 将Message加入到消息队列中
• removeMessages 从队列中移除Message

先看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 (;;) { // 找到Message的插入位置
                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;
}

其中的mMessage相当于队列的头指针，而重点在于理解如何把Message插入到队列中的合适位置。

next方法很长，但做的事主要是去遍历消息队列，找出当前时间可以执行的Message。如队列空了，就阻塞；如果下一个Message的执行时间还未到，则会等待nextPollTimeoutMillis的时间再取出执行。

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

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

removeMessages方法比较简单，分为两步处理：

• 移除消息头中所有符合的Message，mMessage指针也要跟着移动。
• 遍历剩下的消息队列找出所有符合的Message，并移除。

以其中一个为例：

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