Android Looper Handler MessageQueue分析


代码示例来自github上android源码的master分支,随android版本演进实现可能会变化.

looper handler解决的是多线程之间通信的问题,简化了一个具体的模型,通过Message数据结果在不同线程之间传递数据. 工作线程有一个死循环(Looper.loop())不断从MessageQueue中取出其他线程发给自己的消息然后处理.

Message 结构

Message.java

主要成员如下

public final class Message implements Parcelable {
    public int arg1; 
    public int arg2;
    public Object obj;
    int flags;
    long when;
    Bundle data;
    Handler target;
    Runnable callback;
    
    Message next;
    private static Message sPool;
    private static int sPoolSize = 0;
    private static final int MAX_POOL_SIZE = 50;
    
    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 void sendToTarget() {
        target.sendMessage(this);
    }
}

Message通过obtain系列的方法提供了重用Message对象的机制,通过sPool next管理一个缓存链表,最大会缓存MAX_POOL_SIZE = 50个对象,加快执行速度.

Handler

Handler.java

public class Handler {

    final Looper mLooper;
    final MessageQueue mQueue;
    final Callback mCallback;
    
    public interface Callback {
        public boolean handleMessage(Message msg);
    }
    /**
     * Subclasses must implement this to receive messages.
     */
    public void handleMessage(Message msg) {
    }
    
    /**
     * Handle system messages here.
     */
    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();
    }
    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;
    }
    public final boolean sendMessageDelayed(Message msg, long delayMillis)
    {
        return sendMessageAtTime(msg, SystemClock.uptimeMillis() + delayMillis);
    }
    public boolean sendMessageAtTime(Message msg, long uptimeMillis) {
        return enqueueMessage(mQueue, msg, uptimeMillis);
    }
    private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {
        msg.target = this;
        return queue.enqueueMessage(msg, uptimeMillis);
    }
}

handler是对looper的一个封装,主要是把消息enqueueMessage到Looper的MessageQueue中.Looper应该怎样处理消息的回调也会handler中定义. 三种实现方式,一个在Message对象中添加callback(runable)对象,另一是赋值Handler的mCallback成员实现handleMessage方法,另一个是覆写handler的handleMessage方法. 重点关注dispatchMessage这个方法,这是真正干活的函数,搞清这个函数在哪个线程执行很重要.

Looper

Looper.java

public final class Looper {
    static final ThreadLocal<Looper> sThreadLocal = new ThreadLocal<Looper>();
    private static Looper sMainLooper;  // guarded by Looper.class

    final MessageQueue mQueue;
    final Thread mThread;
    
    private static void prepare(boolean quitAllowed) {
        sThreadLocal.set(new Looper(quitAllowed));
    }
    public static void prepareMainLooper() {
        prepare(false);
        synchronized (Looper.class) {
            sMainLooper = myLooper();
        }
    }
    public static Looper myLooper() {
        return sThreadLocal.get();
    }
    private Looper(boolean quitAllowed) {
        mQueue = new MessageQueue(quitAllowed);
        mThread = Thread.currentThread();
    }
    public static void loop() {
        final Looper me = myLooper();
        final MessageQueue queue = me.mQueue;
        for (;;) {
            Message msg = queue.next(); // might block
            try {
                msg.target.dispatchMessage(msg);
            } finally {
            }
            msg.recycleUnchecked();
        }
    }
}

几个点: 1. Looper是怎么和线程结合的,static final ThreadLocal线程专属静态变量,保证线程专属唯一. 2. MainLooper实现和普通的Looper是一样的,是通过synchronized (Looper.class)保证进程唯一同步.文档标明用户不要自己调用prepareMainLooper,系统来调用的. 3. loop()是跑在当前线程的,通过msg.target.dispatchMessage使msg中的回调也是运行在looper所在的线程.

MessageQueue

MessageQueue.java

public final class MessageQueue {
    private long mPtr; // used by native code
    Message mMessages;
    private final ArrayList<IdleHandler> mIdleHandlers = new ArrayList<IdleHandler>();
    private SparseArray<FileDescriptorRecord> mFileDescriptorRecords;
    private IdleHandler[] mPendingIdleHandlers;
    
    // Indicates whether next() is blocked waiting in pollOnce() with a non-zero timeout.
    private boolean mBlocked;
    boolean enqueueMessage(Message msg, long when) {
        synchronized (this) {
            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;
    }
    Message next() {
        int pendingIdleHandlerCount = -1; // -1 only during first iteration
        int nextPollTimeoutMillis = 0;
        for (;;) {
            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;
                }
            }

        }
    }
}

messageQueue使用了poll的方式实现了按时间先后排序的队列,looper在调用next的时候可能被阻塞,等到有message的时间到了才会返回.

运行分析

class LooperThread extends Thread {
    public Handler mHandler;

    public void run() {
        Looper.prepare();

        mHandler = new Handler() {
            public void handleMessage(Message msg) {
                // process incoming messages here
            }
        };

        Looper.loop();
    }
}

结合Handler的处理 java public void dispatchMessage(Message msg) { if (msg.callback != null) { handleCallback(msg); } else { if (mCallback != null) { if (mCallback.handleMessage(msg)) { return; } } handleMessage(msg); } } 上面这种处理走的是第十行的处理.

AsyncTask中对looper和handler的使用分析

AsyncTask文档

基本用法

 private class DownloadFilesTask extends AsyncTask&lt;URL, Integer, Long&gt; {
    protected void onPreExecute() {
    }
     protected Long doInBackground(URL... urls) {
         int count = urls.length;
         long totalSize = 0;
         for (int i = 0; i < count; i++) {
             totalSize += Downloader.downloadFile(urls[i]);
             publishProgress((int) ((i / (float) count) * 100));
             // Escape early if cancel() is called
             if (isCancelled()) break;
         }
         return totalSize;
     }

     protected void onProgressUpdate(Integer... progress) {
         setProgressPercent(progress[0]);
     }

     protected void onPostExecute(Long result) {
         showDialog("Downloaded " + result + " bytes");
     }
 }

调用

new DownloadFilesTask().execute(url1, url2, url3);

AsyncTask.java

public abstract class AsyncTask&lt;Params, Progress, Result&gt; {
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    // We want at least 2 threads and at most 4 threads in the core pool,
    // preferring to have 1 less than the CPU count to avoid saturating
    // the CPU with background work
    private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
    private static final int KEEP_ALIVE_SECONDS = 30;
    private static InternalHandler sHandler;
    
    private final WorkerRunnable<Params, Result> mWorker;
    private final FutureTask<Result> mFuture;
    
    public AsyncTask() {
        mWorker = new WorkerRunnable<Params, Result>() {
            public Result call() throws Exception {
                try {
                    result = doInBackground(mParams);
                } catch (Throwable tr) {
                } finally {
                    postResult(result);
                }
                return result;
            }
        };

        mFuture = new FutureTask<Result>(mWorker) {
            @Override
            protected void done() {
                try {
                    postResultIfNotInvoked(get());
                } catch (InterruptedException e) {
                }
            }
        };
    }

    public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec,
            Params... params) {
        onPreExecute();
        mWorker.mParams = params;
        exec.execute(mFuture);
        return this;
    }
    
    protected final void publishProgress(Progress... values) {
        if (!isCancelled()) {
            getHandler().obtainMessage(MESSAGE_POST_PROGRESS,
                    new AsyncTaskResult&lt;Progress&gt;(this, values)).sendToTarget();
        }
    }
    private Result postResult(Result result) {
        Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
                new AsyncTaskResult<Result>(this, result));
        message.sendToTarget();
        return result;
    }
    //向ui thread投递消息
    private static class InternalHandler extends Handler {
        public InternalHandler() {
            super(Looper.getMainLooper());
        }
        @Override
        public void handleMessage(Message msg) {
            AsyncTaskResult<?> result = (AsyncTaskResult<?>) msg.obj;
            switch (msg.what) {
                case MESSAGE_POST_RESULT:
                    // There is only one result
                    result.mTask.finish(result.mData[0]);
                    break;
                case MESSAGE_POST_PROGRESS:
                    result.mTask.onProgressUpdate(result.mData);
                    break;
            }
        }
    }
    private static class AsyncTaskResult<Data> {
        final AsyncTask mTask;
        final Data[] mData;
        AsyncTaskResult(AsyncTask task, Data... data) {
            mTask = task;
            mData = data;
        }
    }
}

从实现的代码里可以得到一些有益的Looper-handler的用法,还有线程池,executor的用法等.但线程通信的基本原理还是looper-handler的架构.


Copyright © FengGuangtu 2017