Android消息队列原理

本文章是对任玉刚前辈的《Android开发艺术》一书中的第10章“Android的消息机制”的简单理解，不足之处请多多指正。

Handler是Android机制的上层接口，它的运行要依靠MessQueue和Looper。Handler的使用想必大家都很了解，一般在开发中，我们会在子线程中执行耗时的操作，将操作的结果通过Handler发送给主线程，主线程用这个结果来执行UI的操作，这是我们最常见的用法，Android默认只有主线程才能更新UI，这是因为每次更新UI时都会做UI验证操作，Android在UI验证操作时首先会检查当前线程是否为主线程，如果不是就会报出异常，那为什么Android要规定只有主线程才能操作UI呢？这是因为Android的UI控件并不是线程安全的，在高并发状态下当有多个线程访问一个UI时就会出错，加锁又会让UI效率变慢。MessageQueue是消息队列，它以队列的形式对外提供插入和删除消息的工作，但其本身的数据结构并不是一个队列而是一个单向链表。Looper可以理解为消息循环处理器，它会以无限循环的方式去查找MessageQueue中的消息，如果没有消息就会一直等待。Android消息队列的Handler、MessageQueue、Looper作为一个整体，不可分割，那么接下来就对这几个模板分开探索一下。

MessageQueue

MessageQueue即消息队列，主要包含插入(enqueueMessage方法)和读取(next方法)，读取一条消息的同时也会把这条消息从队列中删除，消息队列由单链表实现，因为单链表对插入和删除操作有很好的优势。

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

    }
 

当新消息到来时，将消息插入链表中。

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;

        }

    }
 

next方法会一直阻塞，当有消息到来时，next方法会返回这条消息，并把这条消息从MessageQueue中删除。

Looper

Looper是一个消息循环处理器，在一个线程中，要打开一个Looper才能接收到其他线程发来的Message。在一个线程中，Looper本身是默认不存在的，只有子线程会初始化一个Looper，这就是为什么主线程可以默认使用Handler的原因了。 我们一般用Looper.prepare()方法给线程创建一个Looper，然后用Looper.loop()方法开启消息循环，当然Looper也可以退出，有quit()方法和quitSafely()方法，quit()方法会立即退出这个消息循环，而quitSafely()会将消息队列中的消息处理完再退出。子线程在开启消息循环处理完消息之后一定要退出，否则子线程会一直等待下去，消耗资源。

Looper.loop()方法

    public static void loop() {
final Looper me = myLooper();
if (me == null) {
throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");

        }
if (me.mInLoop) {

            Slog.w(TAG, "Loop again would have the queued messages be executed"

                    + " before this one completed.");

        }


        me.mInLoop = true;
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();

// Allow overriding a threshold with a system prop. e.g.
// adb shell 'setprop log.looper.1000.main.slow 1 && stop && start'
final int thresholdOverride =

                SystemProperties.getInt("log.looper."

                        + Process.myUid() + "."

                        + Thread.currentThread().getName()

                        + ".slow", 0);

boolean slowDeliveryDetected = false;
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
final Printer logging = me.mLogging;
if (logging != null) {

                logging.println(">>>>> Dispatching to " + msg.target + " " +

                        msg.callback + ": " + msg.what);

            }
// Make sure the observer won't change while processing a transaction.
final Observer observer = sObserver;
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;

            Object token = null;
if (observer != null) {

                token = observer.messageDispatchStarting();

            }
long origWorkSource = ThreadLocalWorkSource.setUid(msg.workSourceUid);
try {

                msg.target.dispatchMessage(msg);
if (observer != null) {

                    observer.messageDispatched(token, msg);

                }

                dispatchEnd = needEndTime ? SystemClock.uptimeMillis() : 0;

            } catch (Exception exception) {
if (observer != null) {

                    observer.dispatchingThrewException(token, msg, exception);

                }
throw exception;

            } finally {

                ThreadLocalWorkSource.restore(origWorkSource);
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)) {
// Once we write a slow delivery log, suppress until the queue drains.

                        slowDeliveryDetected = true;

                    }

                }

            }
if (logSlowDispatch) {

                showSlowLog(slowDispatchThresholdMs, dispatchStart, dispatchEnd, "dispatch", 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();

        }

    }
 

loop()方法是一个死循环，只有当消息队列next()方法返回null或者quit()方法被调用后才会跳出死循环，loop()会一直调用next()方法。

Handler

Handler的主要工作就是接收和发送消息，消息发送最终会使用send的一系列方法来实现，在平常使用中我们一般会使用sendMessage()方法发送Message，Handler会调用dispatchMessage()方法。

    public void dispatchMessage(@NonNull Message msg) {
if (msg.callback != null) {

            handleCallback(msg);

        } else {
if (mCallback != null) {
if (mCallback.handleMessage(msg)) {
return;

                }

            }

            handleMessage(msg);

        }

    }
 

最后重写handleMessage方法来接收消息。