異步消息處理線程
對(duì)于普通的線程而言，執(zhí)行完run()方法內(nèi)的代碼后線程就結(jié)束混巧。而異步消息處理線程是指枪向，線程啟動(dòng)后會(huì)進(jìn)入一個(gè)無限循環(huán)體之中，每循環(huán)一次咧党，從其內(nèi)部的消息隊(duì)列中取出一個(gè)消息秘蛔，并回調(diào)相應(yīng)的消息處理函數(shù)，執(zhí)行完一個(gè)消息后繼續(xù)循環(huán)傍衡。如果消息列表為空深员，線程會(huì)暫停，直到消息隊(duì)列中有新的消息蛙埂。
異步線程的實(shí)現(xiàn)思路
- 每個(gè)異步線程內(nèi)部包含一個(gè)消息隊(duì)列辨液，隊(duì)列中的消息一般采用排隊(duì)機(jī)制，先到達(dá)的消息會(huì)先處理箱残。
- 線程的執(zhí)行體中使用while(true)進(jìn)行無限循環(huán)滔迈，循環(huán)體中從消息隊(duì)列中取出消息止吁，并根據(jù)消息的來源，回調(diào)響應(yīng)的消息處理函數(shù)燎悍。
- 其他外部線程可以向本線程的消息隊(duì)列發(fā)送消息敬惦，消息隊(duì)列內(nèi)部的讀寫操作必須進(jìn)行加鎖，不能同時(shí)進(jìn)行讀/寫操作谈山。
Android 消息機(jī)制
Android消息機(jī)制主要指Handler的運(yùn)行機(jī)制俄删，Handler的運(yùn)行機(jī)制需要底層的MessageQueue和Looper的支持。
ThreadLocal
ThreadLocal 是一個(gè)線程內(nèi)部存儲(chǔ)類奏路，通過它可以在指定的線程中存儲(chǔ)數(shù)據(jù)畴椰，同樣只能在指定的線程里獲取存儲(chǔ)的數(shù)據(jù)，其他的線程無法獲取鸽粉。例如對(duì)于Handler來說斜脂，它需要獲取當(dāng)前線程的Looper，Looper的作用域就是線程中触机，并且不同的線程有不同的Looper帚戳。
存儲(chǔ)數(shù)據(jù)
<pre>
public void set(T value) {
Thread currentThread = Thread.currentThread();
Values values = values(currentThread);
if (values == null) {
values = initializeValues(currentThread);
}
values.put(this, value);
}
Values values(Thread current) {
return current.localValues;
}
Values initializeValues(Thread current) {
return current.localValues = new Values();
}
</pre>
意思是先去獲取當(dāng)前線程的Values，null話會(huì)新建一個(gè)儡首。這里這個(gè)Values是Thread的一個(gè)內(nèi)部類片任，這個(gè)內(nèi)部類里有一個(gè)數(shù)組
<pre>
/**
* Map entries. Contains alternating keys (ThreadLocal) and values.
* The length is always a power of 2.
*/
private Object[] table;
</pre>
注釋說存儲(chǔ)交替的key(ThreadLocal的引用)和values，這個(gè)數(shù)組的長(zhǎng)度始終是2的冪蔬胯。
然后是values.put
<pre>
void put(ThreadLocal<?> key, Object value) {
cleanUp();
// Keep track of first tombstone. That's where we want to go back
// and add an entry if necessary.
int firstTombstone = -1;
for (int index = key.hash & mask;; index = next(index)) {
Object k = table[index];
if (k == key.reference) {
// Replace existing entry.
table[index + 1] = value;
return;
}
if (k == null) {
if (firstTombstone == -1) {
// Fill in null slot.
table[index] = key.reference;
table[index + 1] = value;
size++;
return;
}
// Go back and replace first tombstone.
table[firstTombstone] = key.reference;
table[firstTombstone + 1] = value;
tombstones--;
size++;
return;
}
// Remember first tombstone.
if (firstTombstone == -1 && k == TOMBSTONE) {
firstTombstone = index;
}
}
}
</pre>
這里所做的操作驗(yàn)證了剛才的注釋对供，put時(shí)ThreadLocal reference的位置始終在value的前一個(gè)，這樣就把這種鍵值對(duì)交替的存在Thread.values里的Object[] table這個(gè)數(shù)組里氛濒。這樣一方面也說明了一個(gè)Thread里可以存多個(gè)ThreadLocal 和 Value的組合产场。
獲取數(shù)據(jù)
<pre>
public T get() {
// Optimized for the fast path.
Thread currentThread = Thread.currentThread();
Values values = values(currentThread);
if (values != null) {
Object[] table = values.table;
int index = hash & values.mask;
if (this.reference == table[index]) {
return (T) table[index + 1];
}
} else {
values = initializeValues(currentThread);
}
return (T) values.getAfterMiss(this);
}
</pre>
和存相反，這里先通過當(dāng)前ThreadLocal的引用獲取當(dāng)前ThreadLocal在table數(shù)組的位置泼橘，那么他的下一個(gè)位置就是之前存入的數(shù)據(jù)的位置涝动，這樣就可以獲取數(shù)據(jù)了迈勋。
總結(jié)一下：一個(gè)Thread里的Values包含一個(gè)Obj數(shù)組炬灭，這個(gè)數(shù)組的存取是通過ThreadLocal這個(gè)類，存時(shí)用ThreadLocal的引用作為key靡菇，要存的數(shù)據(jù)做value重归，取同樣，多個(gè)ThreadLocal存如同一個(gè)Thread的Values的obj數(shù)組厦凤，獲取數(shù)據(jù)互不影響鼻吮。驗(yàn)證了ThreadLocal的存取操作僅限于線程里。
MessageQueue
MessageQueue消息隊(duì)列主要包含兩個(gè)操作：插入和讀取较鼓。
插入：
<pre>
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("MessageQueue", 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;
}
</pre>
這個(gè)方法事件就是根據(jù)時(shí)間做入隊(duì)的操作椎木。
在hanlder里發(fā)送消息违柏，最后都走了這個(gè)方法：
<pre>
public boolean sendMessageAtTime(Message msg, long uptimeMillis)
{
boolean sent = false;
MessageQueue queue = mQueue;
if (queue != null) {
msg.target = this;
sent = queue.enqueueMessage(msg, uptimeMillis);
}
else {
RuntimeException e = new RuntimeException(
this + " sendMessageAtTime() called with no mQueue");
Log.w("Looper", e.getMessage(), e);
}
return sent;
}
</pre>
可以看見這個(gè)方法都是調(diào)用了MessageQueue的enqueueMessage方法，其中msg參數(shù)就是我們發(fā)送的Message對(duì)象香椎，而uptimeMillis參數(shù)則表示發(fā)送消息的時(shí)間漱竖，它的值等于自系統(tǒng)開機(jī)到當(dāng)前時(shí)間的毫秒數(shù)再加上延遲時(shí)間，如果你調(diào)用的不是sendMessageDelayed()方法畜伐，延遲時(shí)間就為0馍惹，然后將這兩個(gè)參數(shù)都傳遞到MessageQueue的enqueueMessage()方法中。
然后是讀取的next方法：
<pre>
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 (false) Log.v("MessageQueue", "Returning message: " + msg);
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("MessageQueue", "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;
}
}
</pre>
這個(gè)方法非常的長(zhǎng)玛界，還調(diào)用了ndk万矾，大體的意思就是從消息隊(duì)列里不停的讀Message，當(dāng)前時(shí)間大于Message的when就會(huì)返回這個(gè)Message慎框，否則一直會(huì)阻塞良狈，跳出阻塞的條件是：
<pre>
if (mQuitting) {
dispose();
return null;
}
</pre>
Looper
Looper作用是不停地從MessageQueue中查看是否有消息，有消息就會(huì)立即處理鲤脏。下面看它的兩個(gè)方法们颜。
prepare：
<pre>
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));
}
</pre>
結(jié)合ThreadLocal的分析，這個(gè)prepare的操作實(shí)際就是在當(dāng)前線程中存一個(gè)Looper猎醇。
loop：
<pre>
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();
}
}
</pre>
loop中有個(gè)死循環(huán)去調(diào)用MessgeQueue的next方法窥突，當(dāng)next返回的message為null時(shí)才會(huì)跳出死循環(huán)，結(jié)合對(duì)next分析我們知道硫嘶，MessageQueue的msg為空時(shí)會(huì)一直阻塞阻问，只有if（mQuitting）才會(huì)返回null，這時(shí)Looper也會(huì)跳出死循環(huán)沦疾，我們可以通過Loop的quit方法称近，這個(gè)quit方法會(huì)調(diào)用MessgeQueue的quit給mQuitting設(shè)置為true。如果這個(gè)有message哮塞，會(huì)走msg.target.dispatchMessage(msg)方法刨秆。
我們來看一下這個(gè)target是什么進(jìn)入Message類：
<pre>
/package/ Handler target;
</pre>
這個(gè)target就是hanlder。
Handler
<pre>
public Handler(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 that has not called Looper.prepare()");
}
mQueue = mLooper.mQueue;
mCallback = callback;
mAsynchronous = async;
}
public boolean sendMessageAtTime(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);
}
public void dispatchMessage(Message msg) {
if (msg.callback != null) {
handleCallback(msg);
} else {
if (mCallback != null) {
if (mCallback.handleMessage(msg)) {
return;
}
}
handleMessage(msg);
}
}
</pre>
看了這兩個(gè)方法忆畅，我腦子已經(jīng)有一個(gè)畫面了：
總結(jié)
通過Handler發(fā)送消息時(shí)衡未，new Hanlder時(shí)會(huì)檢查當(dāng)前當(dāng)前線程是否有Looper，沒有就會(huì)報(bào)錯(cuò)
"Can't create handler inside thread that has not called Looper.prepare()");
Looper.prepare事件是利用ThreadLocal給當(dāng)前線程存一個(gè)Looper家凯，并通過Looper.Loop獲取并啟動(dòng)他缓醋。handler發(fā)送的消息實(shí)際最后都會(huì)調(diào)用MessageQueue的enQueue的操作，這里會(huì)按時(shí)間排序绊诲，Looper啟動(dòng)后會(huì)不停的讀取MessgeQueue.next的msg送粱，時(shí)間達(dá)到的Msg就會(huì)被分發(fā),調(diào)用handler.dispatchMessage方法，這里實(shí)際就是去調(diào)用在hanlderMessage()里所編寫的業(yè)務(wù)代碼掂之，這樣就成功的講代碼切換到Looper.pre里賦值的線程中執(zhí)行
Tips
另外除了發(fā)送消息之外抗俄，我們還有以下幾種方法可以在子線程中進(jìn)行UI操作：
- Handler的post()方法
- View的post()方法
- Activity的runOnUiThread()方法
