Android的异步消息处理机制
2016-06-05 本文已影响238人
yjiyjige
异步消息处理线程的一般思路
要实现一个异步消息处理线程需要解决如下问题：
- 每个线程应该有一个消息队列，用于对消息进行排队
- 线程执行体中有一个无限的循环，不断地从消息队列中取出消息，并根据消息的来源，去调用相应的处理方法
- 其他线程可以给队列添加消息
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;
}
}
}
