android 进程/线程管理（四）----消息机制的思考（自定义消息机制）

关于android消息机制 已经写了3篇文章了，想要结束这个系列，总觉得少了点什么？

于是我就在想，android为什么要这个设计消息机制，使用消息机制是现在操作系统基本都会有的特点。

可是android是把消息自己提供给开发者使用！我们可以很简单的就在一个线程中创建一个消息系统，不需要考虑同步，消息队列的存放，绑定。

自己搞一个消息系统麻烦吗？android到底为什么要这么设计呢？

那我们自己先搞一个消息机制看看，到底是个什么情况？

首先消息肯定需要消息队列：

package com.joyfulmath.androidstudy.thread.messagemachine;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.joyfulmath.androidstudy.TraceLog;

import android.util.AndroidRuntimeException;

/*message queue
*
* */
public class MessageQueue {
BlockingQueue<Message> mQueue = null;
private boolean mQuit = false;
public MessageQueue()
{
mQueue = new LinkedBlockingQueue<Message>();
mQueue.clear();
}

public boolean enqueueMessage(Message msg, long when)
{
TraceLog.i();
if (msg.target == null) {
throw new AndroidRuntimeException("Message must have a target.");
}

try {
mQueue.put(msg);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
TraceLog.i("done");
return true;
}

public Message next()
{
TraceLog.i();
Message msg = null;
if(mQuit)
{
return null;
}

//wait mQueue msg util we can get one.
try {
msg = mQueue.take();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
TraceLog.i(msg.toString());
return msg;
}

public synchronized void quit()
{
mQuit = true;
}
}

这里有个问题，就是消息添加和获取的同步问题，尤其是一开始，消息队列没有消息的时候，获取消息会怎样？

我们稍后来看这个问题，先看消息队列的几个函数：

BlockingQueue<Message> mQueue = null;

这个就是实际队列存放的地方，就是一个普通的queue。

然后就是加入消息和取出消息。

其实这2个操作，一般都不在一个线程内，需要考虑同步问题。

最后是退出函数，注意，quit()是消息机制结束的标志，一但设置，整个线程将结束。

我们在来讲讲：

mQueue.put(msg);
msg = mQueue.take();

put就是把消息放入队列，而take则与一般的queue peek方法不同，如果消息队列为空，这个地方会block住，直到有消息

加入队列为止。其实这里有个问题，如何一直没有消息进入队列的话，线程会一直block住。

下面我们看看消息队列的生存位置---线程。

package com.joyfulmath.androidstudy.thread.messagemachine;

import com.joyfulmath.androidstudy.TraceLog;

public class MessageHandlerThread extends Thread {
private Object objsync = new Object();

private boolean mQuite = false;
private Message msg = null;
ThreadLocal<MessageQueue> mThreadLocakMsgQueue = new ThreadLocal<MessageQueue>();
@Override
public void run() {
TraceLog.d("MessageHandlerThread start running");
prepare();
final MessageQueue mQueue = getMessageQueue();
while(true)
{
synchronized (objsync) {
if(mQuite)
{
TraceLog.i("quite msg queue");
break;
}
}
Message msg = mQueue.next();
TraceLog.i("get next msg:"+msg);
if(msg == null)
{
// No message indicates that the message queue is quitting.
return;
}
msg.target.dispatchHandlerMessage(msg);
}

TraceLog.i("thread is done");
}

private void prepare()
{
if(mThreadLocakMsgQueue.get()!=null)
{
throw new RuntimeException("message queue should only be one for pre thread");
}
mThreadLocakMsgQueue.set(new MessageQueue());
onPrepared();
}

public MessageQueue getMessageQueue()
{
return mThreadLocakMsgQueue.get();
}

public void quit()
{
synchronized (objsync) {
mQuite = true;
getMessageQueue().quit();
}
}

protected void onPrepared()
{

}
}

MessageHandlerThread
如上，我们定义了一个MessageHandlerThread。

关键的run方法：

public void run() {
TraceLog.d("MessageHandlerThread start running");
prepare();
final MessageQueue mQueue = getMessageQueue();
while(true)
{
synchronized (objsync) {
if(mQuite)
{
TraceLog.i("quite msg queue");
break;
}
}
Message msg = mQueue.next();
TraceLog.i("get next msg:"+msg);
if(msg == null)
{
// No message indicates that the message queue is quitting.
return;
}
msg.target.dispatchHandlerMessage(msg);
}

TraceLog.i("thread is done");
}

while（true） ，是的，消息机制一直在运行着，知道quit的时候。

首先是preopare（）;里面有个变量

ThreadLocal<MessageQueue> mThreadLocakMsgQueue = new ThreadLocal<MessageQueue>();

这是一个特殊的变量，也就是说每个线程拥有一方实例，且，各个线程之间的变量是不可见的。

这样我们保证了，每个messagequeue与thread对应。

接下来我们看看MsgHandler，也就是“操作者”

package com.joyfulmath.androidstudy.thread.messagemachine;

import com.joyfulmath.androidstudy.TraceLog;

public abstract class MsgHandler {
/*
* message queue is only one in thread
* */
final MessageQueue mQueue;

public MsgHandler(MessageQueue mQueue)
{
this.mQueue = mQueue;
}

public void dispatchHandlerMessage(Message msg)
{
TraceLog.i();
onHandleMessage(msg);
}

public void sendMessage(Message msg)
{
enqueueMessage(msg,0L);
}

private boolean enqueueMessage(Message msg, long when)
{
msg.target = this;
return mQueue.enqueueMessage(msg, 0);
}

protected abstract void onHandleMessage(Message msg);
}

其实handler主要任务是2个，把消息加入队列，异步执行消息操作。

protected abstract void onHandleMessage(Message msg);

这个虚函数就是说，每个handler都需要自己去 “handler”自己发送的消息。

public class Message {
public MsgHandler target;
public int what;
@Override
public String toString() {
return "message:"+what;
}

}

最后是message类，只有很简单的2个变量，第一个是在dispatchmessage的时候，知道派送给那个handler来处理。

第二个是消息区分的标志，当然也可以添加其他的一些变量。

以上就是一个简单的消息机制的代码。所以我们总结下消息机制大概需要这么几个角色。

1.消息

2.消息队列

3.thread，也就是消息机制运行的载体

4.handler，也就是“操作者”。

其实根据《android 进程/线程管理（一）----消息机制的框架》，

android消息机制也与上面类似，当然在细节上，android代码的思想的光辉可以看出来。

所以下面将着重分析android消息系统各个精彩的细节，关于系统框架的介绍，可以看本系列的其他文章。

一下源码分析，是基于andorid4.4的，其他版本的源码可能会有不同，请注意。

我们首先来看MessageQueue，andorid的MQ不是一个队列，居然是一个链表！

使用链表的原因应该是，我们不知道queue的长度大小，理论上是足够大，只要内存允许的话。

还有一个重要原因是：

void removeMessages(Handler h, int what, Object object)
void removeMessages(Handler h, Runnable r, Object object)

它可以快速的删除我们不需要的message。

我们来看messagequeue入列和出列操作：

boolean enqueueMessage(Message msg, long when) {
if (msg.isInUse()) {
throw new AndroidRuntimeException(msg + " This message is already in use.");
}
if (msg.target == null) {
throw new AndroidRuntimeException("Message must have a target.");
}

synchronized (this) {
if (mQuitting) {
RuntimeException e = new RuntimeException(
msg.target + " sending message to a Handler on a dead thread");
Log.w("MessageQueue", e.getMessage(), e);
return false;
}

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 (;;) {
if (nextPollTimeoutMillis != 0) {
Binder.flushPendingCommands();
}

// We can assume mPtr != 0 because the loop is obviously still running.
// The looper will not call this method after the loop quits.
nativePollOnce(mPtr, 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);
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("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;
}
}

入列操作：首先是对一些情况的判断，是否设置了handler，是否已经开始退出消息机制等。

然后是把queue加入队列。这些基本可以理解为是同步的，也就是不会block，应为这个操作有极大的可能运行在主线程。

而next恰恰是messagequeue设计的精髓。

nativePollOnce(mPtr, nextPollTimeoutMillis);

线程将CPU交给其他线程运行，自己进入一个等待状态。出发时机就是，timeout或者等待的事件发生了。

所以这个函数是next方法实现等到新消息到来的关键。应此我们可以回答在第一篇里我们提出的那个问题？

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

一开始队列是空的，但是next方法会block住，不会返回null，只有当设置了quit flag以后，才会返回null。

我们看到的enqueueMessage里面的nativeWake就是叫醒这个地方的。

接下去就是取得消息，或者消息时间还没有到，就在此进入等待状态。

当然如果有idehandler需要执行的话，可以执行。

本文实现了一个自定义的消息机制，以及分析了android messagequeue的源码。

关于handler，looper的进一步分析将在下一篇中介绍。