Android消息处理零散分析


       Android消息机制是开发人员用滥的内容,不过内部原理比较复杂,因此今天抽空从头到位彻底分析一下。

前言

       我们知道,Android应用程序是通过消息来驱动的。Google参考了Windows的消息处理机制,在Android系统中实现了一套类似的消息处理机制。
       了解过Win32开发的同学应该都知道windows是事件驱动的,事件驱动围绕着消息的产生与处理展开,事件驱动是靠消息循环机制来实现的。消息机制的三大要点:消息队列、消息循环(分发)、消息处理。其结构如下:

  • 消息队列就是存放消息的一种队列,具有先进先出的特点。每产生一个消息都会添加进消息队列中,在Windows中消息队列是在操作系统中定义的。消息队列就如同一群排队打饭的同学,这群人中光景较好的排在前面,光景较差的排在后面,可以理解成是一种优先级队列!
  • 消息循环就是通过循环(如while)不断地从消息队列中取得队首的消息,并将消息分发出去。类似于食堂打饭的阿姨。
  • 消息处理就是在接收到消息之后根据不同的消息类型做出不同的处理。食堂阿姨根据学生点的不同类型的菜名给他们打不同的饭菜就是消息处理,学生手点的菜名就是消息所携带的信息
  • 事件是根据接收到的消息的具体信息做出的特定的处理,放在代码中是事件响应函数。上面的例子中学生拿到饭菜后吃饭就是具体的事件。

       Android的消息处理机制也是基于上述模型的,实现消息驱动对应模型为:

  • Message:消息,理解为线程间通讯的数据单元。
  • MessageQueue:消息队列,用来存放通过Handler发布的消息,按照先进先出执行。
  • Looper:消息循环,扮演MessageQueue和Handler之间桥梁的角色,循环取出MessageQueue里面的Message,并交付给相应的Handler进行处理。
  • Handler:Handler是Message的主要处理者,负责将Message添加到消息队列以及对消息队列中的Message进行处理。

流程分析

初始化消息队列

       Android应用程序进程在启动时,会在进程中加载ActivityThread类,并且执行这个类的main方法,应用程序的消息循环过程就是在这个main方法里面实现的。ActivityThread位于frameworks/base/core/java/android/app/ActivityThread.java:

1
2
3
4
5
6
7
8
9
10
11
12
13
public final class ActivityThread {
public static void main(String[] args) {
......
Looper.prepareMainLooper();

ActivityThread thread = new ActivityThread();
thread.attach(false);

......
Looper.loop();
......
}
}

       main方法做了两件事,一是在主线程创建了一个ActivityThread实例,二是通过Looper类使主线程进入消息循环。我们只关注后者。先看Looper.prepareMainLooper方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

static final ThreadLocal<Looper> sThreadLocal = new ThreadLocal<Looper>();
private static Looper sMainLooper;
final MessageQueue mQueue;//消息队列
final Thread mThread;//当前线程,这里是主线程

public static void prepareMainLooper() {
prepare(false);//调用prepare(false)方法
synchronized (Looper.class) {
if (sMainLooper != null) {
throw new IllegalStateException("The main Looper has already
been prepared.");
}
sMainLooper = myLooper();
}
}
//带参数prepare方法,参数quitAllowed表示是否可以退出消息循环
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));//存入sThreadLocal
线程局部变量
}
private Looper(boolean quitAllowed) {
mQueue = new MessageQueue(quitAllowed);//初始化一个消息队列
mThread = Thread.currentThread();
}
public static Looper myLooper() {
return sThreadLocal.get();
}

       prepareMainLooper方法其实就是在线程中创建了一个Looper对象,这个Looper对象存放在类型为ThreadLocal的sThreadLocal线程局部变量中,保证每一个调用prepareMainLooper方法的线程只有一个独立的Looper对象。
       Looper对象的构造方法里又创建了一个消息队列MessageQueue,后续消息Message就是存放在这个队列中。我们看看它的创建过程:

1
2
3
4
5
6
7
8
9
10
// True if the message queue can be quit.
private final boolean mQuitAllowed;

private long mPtr; // used by native code

MessageQueue(boolean quitAllowed) {//quitAllowed为false,说明main线程消息循环不允许退出
mQuitAllowed = quitAllowed;
mPtr = nativeInit();
}
private native static long nativeInit();

       MessageQueue的初始化都交给了nativeInit这个本地方法,我们看看它的具体实现,位于frameworks/base/core/jni/android_os_MessageQueue.cpp文件中:

1
2
3
4
5
6
7
8
9
10
static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) {
NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue();//建立一个NativeMessageQueue消息队列对象
if (!nativeMessageQueue) {
jniThrowRuntimeException(env, "Unable to allocate native queue");
return 0;
}

nativeMessageQueue->incStrong(env);//强引用计数+1
return reinterpret_cast<jlong>(nativeMessageQueue);//强转为java层long类型变量,返回给java层mPtr变量保存
}

       在JNI层也相应建立一个消息队列NativeMessageQueue,建立成功后会将它的强引用计数加一,Google的智能指针需要这么干;然后通过C++操作符reinterpret_cast将NativeMessageQueue类型变量强转为jlong类型,并返回到java层,方便java层操作。 NativeMessageQueue同样定义在frameworks/base/core/jni/android_os_MessageQueue.cpp中,我们看看它的构造函数:

1
2
3
4
5
6
7
NativeMessageQueue::NativeMessageQueue() : mInCallback(false), mExceptionObj(NULL) {
mLooper = Looper::getForThread();
if (mLooper == NULL) {
mLooper = new Looper(false);//创建一个JNI层的Looper对象
Looper::setForThread(mLooper);
}
}

       它主要就是在内部创建了一个Looper对象,注意,这个Looper对象是实现在JNI层的,它与上面Java层中的Looper是不一样的,不过它们是对应的,下面我们进一步分析消息循环的过程的时候,就会清楚地了解到它们之间的关系。我们接着看JNI层Looper的构造函数实现,位于system/core/libutils/Looper.cpp中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
Looper::Looper(bool allowNonCallbacks) :
mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
int wakeFds[2];//准备两个文件描述符
int result = pipe(wakeFds);//创建一个管道
LOG_ALWAYS_FATAL_IF(result != 0, "Could not create wake pipe. errno=%d", errno);

mWakeReadPipeFd = wakeFds[0];//管道读端
mWakeWritePipeFd = wakeFds[1];//管道写端

result = fcntl(mWakeReadPipeFd, F_SETFL, O_NONBLOCK);//将管道读端设为非阻塞模式
LOG_ALWAYS_FATAL_IF(result != 0, "Could not make wake read pipe non-blocking. errno=%d",
errno);

result = fcntl(mWakeWritePipeFd, F_SETFL, O_NONBLOCK);//管道写端同样设为非阻塞
LOG_ALWAYS_FATAL_IF(result != 0, "Could not make wake write pipe non-blocking. errno=%d",
errno);

mIdling = false;

// Allocate the epoll instance and register the wake pipe.
mEpollFd = epoll_create(EPOLL_SIZE_HINT);//创建一个epoll专用的文件描述符
LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance. errno=%d", errno);

//epoll其中一个专用结构体
struct epoll_event eventItem;
//把结构体清零
memset(& eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union
//重新赋值
eventItem.events = EPOLLIN;//EPOLLIN :表示对应的文件描述符可以读;
eventItem.data.fd = mWakeReadPipeFd;//fd:关联的文件描述符;
//epoll_ctl函数用于控制某个epoll文件描述符上的事件,可以注册事件,修改事件,删除事件。这里是添加事件
result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeReadPipeFd, & eventItem);
LOG_ALWAYS_FATAL_IF(result != 0, "Could not add wake read pipe to epoll instance. errno=%d",
errno);
}

       这个构造函数做的事情非常重要,它跟我们后面要介绍的应用程序主线程在消息队列中没有消息时要进入等待状态以及当消息队列有消息时要把应用程序主线程唤醒的这两个知识点息息相关。
       它主要做了两件事:
       (1)通过pipe系统调用来创建了一个管道了:

1
2
3
4
5
6
int wakeFds[2];  
int result = pipe(wakeFds);
......

mWakeReadPipeFd = wakeFds[0];
mWakeWritePipeFd = wakeFds[1];

       管道是Linux系统中的一种进程间通信机制。简单来说,管道就是一个文件,在管道的两端,分别是两个打开文件文件描述符,这两个打开文件描述符都是对应同一个文件,其中一个是用来读的,另一个是用来写的。一般的使用方式就是,一个线程通过读文件描述符中来读管道的内容,当管道没有内容时,这个线程就会进入等待状态;而另外一个线程通过写文件描述符来向管道中写入内容,写入内容的时候,如果另一端正有线程正在等待管道中的内容,那么这个线程就会被唤醒。这个等待和唤醒的操作是如何进行的呢,这就要借助Linux系统中的epoll机制了。
       Linux系统中的epoll机制为处理大批量句柄而作了改进的poll,是Linux下多路复用I/O接口select/poll的增强版本,它能显著减少程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。但是这里我们其实只需要监控的I/O接口只有mWakeReadPipeFd一个,即前面我们所创建的管道的读端,为什么还需要用到epoll呢?有点用牛刀来杀鸡的味道。其实不然,这个Looper类是非常强大的,它除了监控内部所创建的管道接口之外,还提供了addFd接口供外界面调用,外界可以通过这个接口把自己想要监控的I/O事件一并加入到这个Looper对象中去,当所有这些被监控的I/O接口上面有事件发生时,就会唤醒相应的线程来处理,不过这里我们只关心刚才所创建的管道的I/O事件的发生。

       (2)epoll机制相关初始化:
       要使用Linux系统的epoll机制,首先要通过epoll_create来创建一个epoll专用的文件描述符:

1
mEpollFd = epoll_create(EPOLL_SIZE_HINT);

       epoll_create(int size)函数生成一个epoll专用的文件描述符。它其实是在内核申请一空间,用来存放你想关注的fd上是否发生以及发生了什么事件。size就是你在这个epoll fd上能关注的最大fd数。我们这里传入的二是EPOLL_SIZE_HINT。

       接着还要通过epoll_ctl函数来告诉epoll要监控相应的文件描述符的什么事件:

1
2
3
4
5
6
7
8
9
//epoll其中一个专用结构体
struct epoll_event eventItem;
//把结构体清零
memset(& eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union
//重新赋值
eventItem.events = EPOLLIN;//EPOLLIN :表示对应的文件描述符可以读;
eventItem.data.fd = mWakeReadPipeFd;//fd:关联的文件描述符;
//epoll_ctl函数用于控制某个epoll文件描述符上的事件,可以注册事件,修改事件,删除事件。这里是添加事件
result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeReadPipeFd, & eventItem);

       epoll_ctl函数如下:
函数声明:int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
该函数用于控制某个epoll文件描述符上的事件,可以注册事件,修改事件,删除事件。
参数:
epfd:由 epoll_create 生成的epoll专用的文件描述符;
op:要进行的操作例如注册事件,可能的取值EPOLL_CTL_ADD 注册、EPOLL_CTL_MOD 修 改、EPOLL_CTL_DEL 删除
fd:关联的文件描述符;
event:指向epoll_event结构体的指针;
返回值:
如果调用成功返回0,不成功返回-1

       用到的结构体如下:

1
2
3
4
5
6
7
8
9
10
11
struct epoll_event {   
__uint32_t events; /* Epoll events */ //事件类型
epoll_data_t data; /* User data variable */
};

typedef union epoll_data {
void *ptr;
int fd; //关联的文件描述符
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;

       epoll_event 结构体常用的事件类型:
EPOLLIN :表示对应的文件描述符可以读;
EPOLLOUT:表示对应的文件描述符可以写;
EPOLLPRI:表示对应的文件描述符有紧急的数据可读
EPOLLERR:表示对应的文件描述符发生错误;
EPOLLHUP:表示对应的文件描述符被挂断;
EPOLLET:表示对应的文件描述符有事件发生;

       这里就是告诉mEpollFd,它要监控mWakeReadPipeFd文件描述符的EPOLLIN事件,即当管道中有内容可读时,就唤醒当前正在等待管道中的内容的线程。
       C++层的这个Looper对象创建好了之后,就返回到JNI层的NativeMessageQueue的构造函数,最后就返回到Java层的消息队列MessageQueue的创建过程,这样,Java层的Looper对象就准备好了。

       对上面内容小结一下就是:
       ①在Java层,创建了一个Looper对象,这个Looper对象是用来进入消息循环的,它的内部有一个消息队列MessageQueue对象mQueue;
       ②在JNI层,创建了一个NativeMessageQueue对象,这个NativeMessageQueue对象保存在Java层的消息队列对象mQueue的成员变量mPtr中;
       ③在C++层,创建了一个Looper对象,保存在JNI层的NativeMessageQueue对象的成员变量mLooper中,这个对象的作用是,当Java层的消息队列中没有消息时,就使Android应用程序主线程进入等待状态,而当Java层的消息队列中来了新的消息后,就唤醒Android应用程序的主线程来处理这个消息。

消息循环

       继续回到上面ActivitThread的main方法里,在上面这些工作都准备好之后就调用Looper.loop方法进入到消息循环中了。
       消息循环就会取出消息进行处理,在看消息处理之前,先看一下消息是怎么被添加到消息队列的。

发送消息

       在Java层,Message类表示一个消息对象,要发送消息首先就要先获得一个消息对象,Message类的构造函数是public的,但是不建议直接new Message,Message内部保存了一个缓存的消息池,我们可以用obtain从缓存池获得一个消息,Message使用完后系统会调用recycle回收,如果自己new很多Message,每次使用完后系统放入缓存池,会占用很多内存的。我们看看Message类相关方法和变量:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
private static Message sPool;
// sometimes we store linked lists of these things
/*package*/ Message next;
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 recycle() {
if (isInUse()) {
if (gCheckRecycle) {
throw new IllegalStateException("This message cannot be recycled because it "
+ "is still in use.");
}
return;
}
recycleUnchecked();
}
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内部通过next成员实现了一个链表,这样sPool就了为了一个Messages的缓存链表。

       消息对象获取到了怎么发送呢,大家都知道是通过Handler的post、sendMessage等方法,其实这些方法最终都是调用的同一个方法sendMessageAtTime:

1
2
3
4
5
6
7
8
9
10
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);
}

       sendMessageAtTime获取到消息队列然后调用enqueueMessage方法,消息队列mQueue是从与Handler关联的Looper获得的。

1
2
3
4
5
6
7
private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {
msg.target = this;
if (mAsynchronous) {
msg.setAsynchronous(true);
}
return queue.enqueueMessage(msg, uptimeMillis);
}

       enqueueMessage将message的target设置为当前的handler,然后调用MessageQueue的enqueueMessage,在调用queue.enqueueMessage之前判断了mAsynchronous,从名字看是异步消息的意思,要明白Asynchronous的作用,需要先了解一个概念Barrier。

Barrier与Asynchronous Message

       Barrier是什么意思呢,从名字看是一个拦截器,在这个拦截器后面的消息都暂时无法执行,直到这个拦截器被移除了,MessageQueue有一个函数叫enqueueSyncBarier可以添加一个Barrier。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
//消息队列的队头,消息队列按时间从小到大排序,这是最小的
Message mMessages;

int enqueueSyncBarrier(long when) {
// Enqueue a new sync barrier token.
// We don't need to wake the queue because the purpose of a barrier is to stall it.
synchronized (this) {
final int token = mNextBarrierToken++;//一个用来返回的自增int值作为token
final Message msg = Message.obtain();//创建一个路障
msg.markInUse();//标记为正在使用
msg.when = when;//路障的时间
msg.arg1 = token;//打个token

Message prev = null;
Message p = mMessages;//访问消息队列队头
if (when != 0) {
while (p != null && p.when <= when) {//遍历一遍消息队列,按时间找到路障位置
prev = p;
p = p.next;
}
}

if (prev != null) { // invariant: p == prev.next //在时间所在队列位置插入路障
msg.next = p;
prev.next = msg;
} else {//路障时间小于队头时间,或者消息队列为空,将路障设为队头
msg.next = p;
mMessages = msg;
}
return token;
}
}

       在enqueueSyncBarrier中,obtain了一个Message,并设置msg.arg1=token,token仅是一个每次调用enqueueSyncBarrier时自增的int值,目的是每次调用enqueueSyncBarrier时返回唯一的一个token,这个Message同样需要设置执行时间,然后插入到消息队列,特殊的是这个Message没有设置target,即msg.target为null。
       如果队列头部的消息的target为null就表示它是个Barrier,因为只有两种方法往mMessages中添加消息,一种是enqueueMessage,另一种是enqueueBarrier,而enqueueMessage中如果mst.target为null是直接抛异常的,后面会看到。

       所谓的异步消息其实就是这样的,我们可以通过enqueueBarrier往消息队列中插入一个Barrier,那么队列中执行时间在这个Barrier以后的同步消息都会被这个Barrier拦截住无法执行,直到我们调用removeBarrier移除了这个Barrier,而异步消息则没有影响,消息默认就是同步消息,除非我们调用了Message的setAsynchronous,这个方法是隐藏的。只有在初始化Handler时通过参数指定往这个Handler发送的消息都是异步的,这样在Handler的enqueueMessage中就会调用Message的setAsynchronous设置消息是异步的,从上面Handler.enqueueMessage的代码中可以看到。
       所谓异步消息,其实只有一个作用,就是在设置Barrier时仍可以不受Barrier的影响被正常处理,如果没有设置Barrier,异步消息就与同步消息没有区别,可以通过removeSyncBarrier移除Barrier:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
void removeSyncBarrier(int token) {
// Remove a sync barrier token from the queue.
// If the queue is no longer stalled by a barrier then wake it.
synchronized (this) {
Message prev = null;
Message p = mMessages;//取出队头
while (p != null && (p.target != null || p.arg1 != token)) {//循环遍历消息队列,找出路障
prev = p;
p = p.next;
}
if (p == null) {//参数token就是enqueueSyncBarrier的返回值,如果没有调用指定的token是会抛异常的
throw new IllegalStateException("The specified message queue synchronization "
+ " barrier token has not been posted or has already been removed.");
}
final boolean needWake;
if (prev != null) {//找到路障,移除路障
prev.next = p.next;
needWake = false;
} else {
mMessages = p.next;
needWake = mMessages == null || mMessages.target != null;
}
p.recycleUnchecked();

// If the loop is quitting then it is already awake.
// We can assume mPtr != 0 when mQuitting is false.
if (needWake && !mQuitting) {
nativeWake(mPtr);//唤醒操作下面再讲
}
}
}

enqueueMessage

       接着我们顺着上面内容,看一下是MessageQueue的enqueueMessage的方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
boolean enqueueMessage(Message msg, long when) {
if (msg.target == null) {//注意这里,当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;
//如果当前的消息队列为空,或者新添加的消息的执行时间when是0,
//或者新添加的消息的执行时间比消息队列头的消息的执行时间还早,
//就把消息添加到消息队列头(消息队列按时间排序)
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;
}

       注意方法内第一行,当msg.target为null时是直接抛异常的。

       在enqueueMessage中首先判断,如果当前的消息队列为空,这时候应用程序的主线程一般就是处于空闲等待状态了,这时候就要唤醒它(唤醒我们下面再讲)。或者新添加的消息的执行时间when是0,或者新添加的消息的执行时间比消息队列头的消息的执行时间还早,就把消息添加到消息队列头(消息队列按时间排序),否则就要找到合适的位置将当前消息添加到消息队列。

消息循环

       消息队列初始化好了,也知道怎么发消息了,下面就是怎么处理消息了,看Looper.loop函数:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
  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();
}
}

       这里就是进入到消息循环中去了,它不断地从消息队列mQueue中去获取下一个要处理的消息msg,如果消息的target成员变量为null,就表示要退出消息循环了,否则的话就要调用这个target对象的dispatchMessage成员函数来处理这个消息,这个target对象的类型为Handler,下面我们分析消息的发送时会看到这个消息对象msg是如设置的。
       这个函数最关键的地方便是从消息队列中获取下一个要处理的消息了,即MessageQueue.next函数,我们看看它的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 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 (;;) {
......
//本地方法,一次消息轮训
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;
}
}

       调用这个函数的时候,有可能会让线程进入等待状态。什么情况下,线程会进入等待状态呢?两种情况,一是当消息队列中没有消息时,它会使线程进入等待状态;二是消息队列中有消息,但是消息指定了执行的时间,而现在还没有到这个时间,线程也会进入等待状态。消息队列中的消息是按时间先后来排序的,上面我们分析过了。

       执行下面语句是看看当前消息队列中有没有消息:

1
nativePollOnce(mPtr, nextPollTimeoutMillis);

       这是一个JNI方法,我们等一下再分析,这里传入的参数mPtr就是指向前面我们在JNI层创建的NativeMessageQueue对象了,而参数nextPollTimeoutMillis则表示如果当前消息队列中没有消息,它要等待的时候,for循环开始时,传入的值为0,表示不等待。
       当前nativePollOnce返回后,就去看看消息队列中有没有消息:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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;
}

       取消息队列头部的消息,如果头部消息是Barrier(target==null)就往后遍历找到第一个异步消息;

       提前说明:nextPollTimeoutMillis 的值的意义是,-1表示下次调用nativePollOnce时,如果消息中没有消息,就进入无限等待状态中去。
       这里计算出来的等待时间都是在下次调用nativePollOnce时使用的。

       接下来检测获取到的消息(消息队列头部的消息或者第一个异步消息),如果为null表示没有消息要执行,设置nextPollTimeoutMillis = -1,就要无限等待下去;
       否则检测这个消息要执行的时间,如果到执行时间了就将这个消息markInUse并从消息队列移除,然后从next返回到loop;
       否则设置nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE),即距离最近要执行的消息还需要多久。

       这里说的等待,是空闲等待,而不是忙等待,因此,在进入空闲等待状态前,如果应用程序注册了IdleHandler接口来处理一些事情,那么就会先执行这里IdleHandler,然后再进入等待状态。IdlerHandler是定义在MessageQueue的一个内部类:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
* Callback interface for discovering when a thread is going to block
* waiting for more messages.
*/

public static interface IdleHandler {
/**
* Called when the message queue has run out of messages and will now
* wait for more. Return true to keep your idle handler active, false
* to have it removed. This may be called if there are still messages
* pending in the queue, but they are all scheduled to be dispatched
* after the current time.
*/

boolean queueIdle();
}

       它只有一个成员函数queueIdle,执行这个函数时,如果返回值为false,那么就会从应用程序中移除这个IdleHandler,否则的话就会在应用程序中继续维护着这个IdleHandler,下次空闲时仍会再执会这个IdleHandler。MessageQueue提供了addIdleHandler和removeIdleHandler两注册和删除IdleHandler。

       回到MessageQueue的next方法中,它接下来就是在进入等待状态前,看看有没有IdleHandler是需要执行的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 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);

       如果没有,即pendingIdleHandlerCount等于0,那下面的逻辑就不执行了,通过continue语句直接进入下一次循环,否则就要把注册在mIdleHandlers中的IdleHandler取出来,放在mPendingIdleHandlers数组中去。

       接下来就是执行这些注册了的IdleHanlder了:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 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);
}
}
}

       执行完这些IdleHandler之后,线程下次调用nativePollOnce函数时,就不设置超时时间了,因为,很有可能在执行IdleHandler的时候,已经有新的消息加入到消息队列中去了,因此,要重置nextPollTimeoutMillis的值:

1
2
3
// 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;

       到这里MessageQueue的next方法的java部分就分析完了,我们重点看一下native方法nativePollOnce,看看它是如何进入等待状态的。这个函数定义在frameworks/base/core/jni/android_os_MessageQueue.cpp文件中:

1
2
3
4
5
static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jclass clazz,
jlong ptr, jint timeoutMillis)
{

NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
nativeMessageQueue->pollOnce(env, timeoutMillis);
}

       这个函数首先是通过传进入的参数ptr取回前面在Java层创建MessageQueue对象时在JNI层创建的NatvieMessageQueue对象,然后调用它的pollOnce函数:

1
2
3
4
5
void NativeMessageQueue::pollOnce(JNIEnv* env, int timeoutMillis) {
......
mLooper->pollOnce(timeoutMillis);
......
}

       这里将操作转发给mLooper对象的pollOnce函数处理,这里的mLooper对象是在C++层的对象,它也是在前面在JNI层创建的NatvieMessageQueue对象时创建的,它的pollOnce函数定义在system/core/libutils/Looper.cpp中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {  
int result = 0;
for (;;) {
......

if (result != 0) {
......

return result;
}

result = pollInner(timeoutMillis);
}
}

       省略一些障眼法,它主要就是调用pollInner函数来进一步操作,如果pollInner返回值不等于0,这个函数就可以返回了。函数pollInner的定义如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
int Looper::pollInner(int timeoutMillis) {  
......

int result = ALOOPER_POLL_WAKE;

......

#ifdef LOOPER_USES_EPOLL
struct epoll_event eventItems[EPOLL_MAX_EVENTS];
int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
bool acquiredLock = false;
#else
......
#endif

if (eventCount < 0) {
if (errno == EINTR) {
goto Done;
}

LOGW("Poll failed with an unexpected error, errno=%d", errno);
result = ALOOPER_POLL_ERROR;
goto Done;
}

if (eventCount == 0) {
......
result = ALOOPER_POLL_TIMEOUT;
goto Done;
}

......

#ifdef LOOPER_USES_EPOLL
for (int i = 0; i < eventCount; i++) {
int fd = eventItems[i].data.fd;
uint32_t epollEvents = eventItems[i].events;
if (fd == mWakeReadPipeFd) {
if (epollEvents & EPOLLIN) {
awoken();
} else {
LOGW("Ignoring unexpected epoll events 0x%x on wake read pipe.", epollEvents);
}
} else {
......
}
}
if (acquiredLock) {
mLock.unlock();
}
Done: ;
#else
......
#endif

......

return result;
}

       省略一些障眼法,这里,首先是调用epoll_wait函数来看看epoll专用文件描述符mEpollFd所监控的文件描述符是否有IO事件发生,它设置监控的超时时间为timeoutMillis:

1
int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);

       epoll_wait函数 如下:
函数声明:int epoll_wait(int epfd,struct epoll_event * events,int maxevents,int timeout)
该函数用于轮询I/O事件的发生;
参数:
epfd:由epoll_create 生成的epoll专用的文件描述符;
epoll_event:用于回传代处理事件的数组;
maxevents:每次能处理的事件数;
timeout:等待I/O事件发生的超时值;-1相当于阻塞,0相当于非阻塞。一般用-1即可
返回值:
返回发生事件数。

       epoll_wait运行的原理是 :
等侍注册在epfd上的fd的事件的发生,如果发生则将发生的fd和事件类型放入到events数组中。
并 且将注册在epfd上的fd的事件类型给清空,所以如果下一个循环你还要关注这个fd的话,则需要用epoll_ctl(epfd,EPOLL_CTL_MOD,listenfd,&ev)来重新设置fd的事件类型。这时不用EPOLL_CTL_ADD,因为fd并未清空,只是事件类型清空。

       回忆一下前面的Looper的构造函数,我们在里面设置了要监控mWakeReadPipeFd文件描述符的EPOLLIN事件。
       当mEpollFd所监控的文件描述符发生了要监控的I/O事件后或者监控时间超时后,线程就从epoll_wait返回了,否则线程就会在epoll_wait函数中进入睡眠状态了。返回后如果eventCount等于0,就说明是超时了:

1
2
3
4
5
if (eventCount == 0) {  
......
result = ALOOPER_POLL_TIMEOUT;
goto Done;
}

        如果eventCount不等于0,就说明发生要监控的事件:

1
2
3
4
5
6
7
8
9
10
11
12
13
for (int i = 0; i < eventCount; i++) {  
int fd = eventItems[i].data.fd;
uint32_t epollEvents = eventItems[i].events;
if (fd == mWakeReadPipeFd) {
if (epollEvents & EPOLLIN) {
awoken();
} else {
LOGW("Ignoring unexpected epoll events 0x%x on wake read pipe.", epollEvents);
}
} else {
......
}
}

       这里我们只关注mWakeReadPipeFd文件描述符上的事件,如果在mWakeReadPipeFd文件描述符上发生了EPOLLIN就说明应用程序中的消息队列里面有新的消息需要处理了,接下来它就会先调用awoken函数清空管道中的内容,以便下次再调用pollInner函数时,知道自从上次处理完消息队列中的消息后,有没有新的消息加进来。
       函数awoken的实现很简单,它只是把管道中的内容都读取出来:

1
2
3
4
5
6
7
8
9
void Looper::awoken() {  
......

char buffer[16];
ssize_t nRead;
do {
nRead = read(mWakeReadPipeFd, buffer, sizeof(buffer));
} while ((nRead == -1 && errno == EINTR) || nRead == sizeof(buffer));
}

       因为当其它的线程向应用程序的消息队列加入新的消息时,会向这个管道写入新的内容来通知应用程序主线程有新的消息需要处理了,从而唤醒它,下面我们分析一下。

唤醒时机

       上面讲发送消息时讲过,把消息加入到消息队列时,分两种情况,一种当前消息队列为空时,这时候应用程序的主线程一般就是处于空闲等待状态了,这时候就要唤醒它,另一种情况是应用程序的消息队列不为空,这时候就不需要唤醒应用程序的主线程了,因为这时候它一定是在忙着处于消息队列中的消息,因此不会处于空闲等待的状态。
       把消息加入到消息队列去后,如果应用程序的主线程正处于空闲等待状态,就需要调用natvieWake函数来唤醒它了,这是一个JNI方法,定义在frameworks/base/core/jni/android_os_MessageQueue.cpp文件中:

1
2
3
4
static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
return nativeMessageQueue->wake();
}

       这个JNI层的NativeMessageQueue对象我们在前面分析消息循环的时候创建好的,保存在Java层的MessageQueue对象的mPtr成员变量中,这里把它取回来之后,就调用它的wake函数来唤醒应用程序的主线程,这个函数也是定义在frameworks/base/core/jni/android_os_MessageQueue.cpp文件中:

1
2
3
void NativeMessageQueue::wake() {
mLooper->wake();
}

       这里它又通过成员变量mLooper的wake函数来执行操作,这里的mLooper成员变量是一个C++层实现的Looper对象,它定义在system/core/libutils/Looper.cpp文件中:

1
2
3
4
5
6
7
8
9
10
void Looper::wake() {  
......

ssize_t nWrite;
do {
nWrite = write(mWakeWritePipeFd, "W", 1);
} while (nWrite == -1 && errno == EINTR);

.......
}

       这个wake函数很简单,只是通过打开文件描述符mWakeWritePipeFd往管道的写入一个”W”字符串。其实,往管道写入什么内容并不重要,往管道写入内容的目的是为了唤醒应用程序的主线程。前面我们在分析应用程序的消息循环时说到,当应用程序的消息队列中没有消息处理时,应用程序的主线程就会进入空闲等待状态,而这个空闲等待状态就是通过调用这个Looper类的pollInner函数来进入的,具体就是在pollInner函数中调用epoll_wait函数来等待管道中有内容可读的。
       这时候既然管道中有内容可读了,应用程序的主线程就会从这里的Looper类的pollInner函数返回到JNI层的nativePollOnce函数,最后返回到Java层中的MessageQueue.next函数中去,这里它就会发现消息队列中有新的消息需要处理了,于就会处理这个消息。

       剩下就是对消息的处理了,这个就更随意了。回到Looper.loop方法中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public class Looper {  
......

public static final void loop() {
Looper me = myLooper();
MessageQueue queue = me.mQueue;

......

while (true) {
Message msg = queue.next(); // might block
......

if (msg != null) {
if (msg.target == null) {
// No target is a magic identifier for the quit message.
return;
}

......

msg.target.dispatchMessage(msg); //处理消息

......

msg.recycle();
}
}
}

......
}

       它从消息队列中获得消息对象msg后,就会调用它的target成员变量的dispatchMessage函数来处理这个消息。在前面分析消息的发送时说过,这个消息对象msg的成员变量target是在发送消息的时候设置好的,一般就通过哪个Handler来发送消息,就通过哪个Handler来处理消息。
       我们这里的Handler是ActivityThread的成员变量mH,是一个类型为H的Handler。 H类没有实现自己的dispatchMessage函数,但是它继承了父类Handler的dispatchMessage函数,我们看看Handler的dispatchMessage方法:

1
2
3
4
5
6
7
8
9
10
11
12
public void dispatchMessage(Message msg) {
if (msg.callback != null) {
handleCallback(msg);
} else {
if (mCallback != null) {
if (mCallback.handleMessage(msg)) {
return;
}
}
handleMessage(msg);
}
}

       这里的消息对象msg的callback成员变量和Handler类的mCallBack成员变量一般都为null,于是,就会调用Handler类的handleMessage函数来处理这个消息,由于H类在继承Handler类时,重写了handleMessage函数,因此,这里调用的实际上是H类的handleMessage函数。

总结

       至此,我们就分析完Android应用程序的消息处理机制了,简单做一个总结:

  • Android应用程序的主线程在进入消息循环过程前,会在内部创建一个Linux管道(Pipe),这个管道的作用是使得Android应用程序主线程在消息队列为空时可以进入空闲等待状态,并且使得当应用程序的消息队列有消息需要处理时唤醒应用程序的主线程。

  • Android应用程序的主线程进入空闲等待状态的方式实际上就是在管道的读端等待管道中有新的内容可读,具体来说就是是通过Linux系统的Epoll机制中的epoll_wait函数进行的。

  • 当往Android应用程序的消息队列中加入新的消息时,会同时往管道中的写端写入内容,通过这种方式就可以唤醒正在等待消息到来的应用程序主线程。

  • 当应用程序主线程在进入空闲等待前,会认为当前线程处理空闲状态,于是就会调用那些已经注册了的IdleHandler接口,使得应用程序有机会在空闲的时候处理一些事情。

       上述分析的是系统启动过程中的消息机制,如果我们想看一下一个线程想实现消息循环应该怎么做,可以看看HandlerThread的实现,和上面内容大同小异,有兴趣的同学可以自己研究研究。

       结尾处贴个妹纸吧,这是今年ChinaJoy会场一个show girl,前同事拍的(我设备太渣了,所以在馆内只参加了活动,玩了些游戏= 。=)。


坚持技术分享,您的支持将鼓励我继续创作!