你的位置:首页 > 操作系统

[操作系统]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的进一步分析将在下一篇中介绍。