在深入解析Android中Handler消息机制一文中，我们学习了Handler消息机制的java层代码，这次我们来学习Handler消息机制的native层代码。
在Java层的消息处理机制中，MessageQueue类里面涉及到多个native方法，除了MessageQueue的native方法，native层本身也有一套完整的消息机制，用于处理native的消息。在整个消息机制中，而MessageQueue是连接Java层和Native层的纽带，换言之，Java层可以向MessageQueue消息队列中添加消息，Native层也可以向MessageQueue消息队列中添加消息。
MessageQueue
MessageQueue是在Looper的构造方法里面创建的
MessageQueue中设计的native方法如下：
private native static long nativeInit();
private native static void nativeDestroy(long ptr);
private native void nativePollOnce(long ptr, int timeoutMillis);
private native static void nativeWake(long ptr);
private native static boolean nativeIsPolling(long ptr);
private native static void nativeSetFileDescriptorEvents(long ptr, int fd, int events);private Looper(boolean quitAllowed) {
mQueue = new MessageQueue(quitAllowed);
mThread = Thread.currentThread();
}
nativeInit()
1. new MessageQueue()
首先，我们从MessageQueue的构造函数入手，其中调用了nativeInit()方法
MessageQueue(boolean quitAllowed) {
mQuitAllowed = quitAllowed;
mPtr = nativeInit(); //mPtr记录native消息队列的信息
}2.android_os_MessageQueue_nativeInit()方法
framework/base/core/jni/android_os_MessageQueue.cpp
static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) {
NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue(); //初始化native消息队列
if (!nativeMessageQueue) {
jniThrowRuntimeException(env, "Unable to allocate native queue");
return 0;
}
nativeMessageQueue->incStrong(env);
return reinterpret_cast<jlong>(nativeMessageQueue);
}3.new NativeMessageQueue()
NativeMessageQueue::NativeMessageQueue() : mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {
mLooper = Looper::getForThread(); //获取TLS中的Looper对象
if (mLooper == NULL) {
mLooper = new Looper(false); //创建native层的Looper
Looper::setForThread(mLooper); //保存native层的Looper到TLS中
}Looper::getForThread()，功能类比于Java层的Looper.myLooper();
Looper::setForThread(mLooper)，功能类比于Java层的ThreadLocal.set();
MessageQueue是在Java层与Native层有着紧密的联系，但是在上面的代码中似乎Native层的Looper与Java层的Looper没有任何的关系，可以发现native基本等价于用C++重写了Java的Looper逻辑，故可以发现很多功能类似的地方。
4.new Looper()
system/core/libutils/Looper.cpp
Looper::Looper(bool allowNonCallbacks) :
mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false),
mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
mWakeEventFd = eventfd(0, EFD_NONBLOCK); //构造唤醒事件的fd
AutoMutex _l(mLock);
rebuildEpollLocked(); //重建Epoll事件
}5.epoll_create/epoll_ctl
void Looper::rebuildEpollLocked() {
if (mEpollFd >= 0) {
close(mEpollFd); //关闭旧的epoll实例
}
mEpollFd = epoll_create(EPOLL_SIZE_HINT); //创建新的epoll实例，并注册wake管道
struct epoll_event eventItem;
memset(& eventItem, 0, sizeof(epoll_event)); //把未使用的数据区域进行置0操作
eventItem.events = EPOLLIN; //可读事件
eventItem.data.fd = mWakeEventFd;
//将唤醒事件(mWakeEventFd)添加到epoll实例(mEpollFd)
int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);
for (size_t i = 0; i < mRequests.size(); i++) {
const Request& request = mRequests.valueAt(i);
struct epoll_event eventItem;
request.initEventItem(&eventItem);
//将request队列的事件，分别添加到epoll实例
int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, & eventItem);
if (epollResult < 0) {
ALOGE("Error adding epoll events for fd %d while rebuilding epoll set, errno=%d", request.fd, errno);
}
}
}关于epoll，此处不展开说明。
此处注意Request队列，也添加到epoll的监控范围内。
Looper.cpp.该类中提供了pollOnce 和wake的休眠和唤醒机制。同时在构造函数中也创建 管道 并加入epoll的机制中，来监听其状态变化。
总结一下，初始化的流程图：
nativeDestroy()
查看了MessageQueue在native层的初始化后，我们来看一下MessageQueue在native层的销毁流程。
MessageQueue.java
private void dispose() {
if (mPtr != 0) {
nativeDestroy(mPtr);
mPtr = 0;
}
}2.android_os_MessageQueue_nativeDestroy()
android_os_MessageQueue.cpp
static void android_os_MessageQueue_nativeDestroy(JNIEnv* env, jclass clazz, jlong ptr) {
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
nativeMessageQueue->decStrong(env);
}nativeMessageQueue继承自RefBase类，所以decStrong最终调用的是RefBase.decStrong().
3.RefBase::decStrong()
system/core/libutils/RefBase.cpp
void RefBase::decStrong(const void* id) const
{
weakref_impl* const refs = mRefs;
refs->removeStrongRef(id); //移除强引用
const int32_t c = android_atomic_dec(&refs->mStrong);
if (c == 1) {
refs->mBase->onLastStrongRef(id);
if ((refs->mFlags&OBJECT_LIFETIME_MASK) == OBJECT_LIFETIME_STRONG) {
delete this;
}
}
refs->decWeak(id); // 移除弱引用
}关于RefBase的更多知识，请看Android Framework学习(六)之RefBase,SP,WP
归纳一下销毁的流程图
nativePollOnce()
nativePollOnce用于提取消息队列中的消息
1.MessageQueue.next()
MessageQueue.java
Looper中的loop()方法会调用MessageQueue的next()方法，不断从MessageQueue中获取Message 然后分发给对应的Handler处理
Message next() {
final long ptr = mPtr;
if (ptr == 0) {
return null;
}
for (;;) {
...
nativePollOnce(ptr, nextPollTimeoutMillis); //阻塞操作
...
}2.android_os_MessageQueue_nativePollOnce()
android_os_MessageQueue.cpp
static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj, jlong ptr, jint timeoutMillis) {
//将Java层传递下来的mPtr转换为nativeMessageQueue
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
nativeMessageQueue->pollOnce(env, obj, timeoutMillis);
}3.NativeMessageQueue::pollOnce()
android_os_MessageQueue.cpp
void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) {
mPollEnv = env;
mPollObj = pollObj;
mLooper->pollOnce(timeoutMillis);
mPollObj = NULL;
mPollEnv = NULL;
if (mExceptionObj) {
env->Throw(mExceptionObj);
env->DeleteLocalRef(mExceptionObj);
mExceptionObj = NULL;
}
}4.Looper::pollOnce()
framework/native/include/android/Looper.h
inline int pollOnce(int timeoutMillis) {
return pollOnce(timeoutMillis, NULL, NULL, NULL);
}5.Looper::pollOnce()
framework/base/native/android/looper.cpp
int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {
int result = 0;
for (;;) {
// 先处理没有Callback方法的 Response事件
while (mResponseIndex < mResponses.size()) {
const Response& response = mResponses.itemAt(mResponseIndex++);
int ident = response.request.ident;
if (ident >= 0) { //ident大于0，则表示没有callback, 因为POLL_CALLBACK = -2,
int fd = response.request.fd;
int events = response.events;
void* data = response.request.data;
if (outFd != NULL) *outFd = fd;
if (outEvents != NULL) *outEvents = events;
if (outData != NULL) *outData = data;
return ident;
}
}
if (result != 0) {
if (outFd != NULL) *outFd = 0;
if (outEvents != NULL) *outEvents = 0;
if (outData != NULL) *outData = NULL;
return result;
}
// 再处理内部轮询
result = pollInner(timeoutMillis);
}
}参数说明：
timeoutMillis：超时时长
outFd：发生事件的文件描述符
outEvents：当前outFd上发生的事件，包含以下4类事件
EVENT_INPUT 可读
EVENT_OUTPUT 可写
EVENT_ERROR 错误
EVENT_HANGUP 中断
outData：上下文数据
6.Looper::pollInner()
Looper.cpp
int Looper::pollInner(int timeoutMillis) {
...
int result = POLL_WAKE;
mResponses.clear();
mResponseIndex = 0;
mPolling = true; //即将处于idle状态
struct epoll_event eventItems[EPOLL_MAX_EVENTS]; //fd最大个数为16
//等待事件发生或者超时，在nativeWake()方法，向管道写端写入字符，则该方法会返回；
int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
mPolling = false; //不再处于idle状态
mLock.lock(); //请求锁
if (mEpollRebuildRequired) {
mEpollRebuildRequired = false;
rebuildEpollLocked(); // epoll重建，直接跳转Done;
goto Done;
}
if (eventCount < 0) {
if (errno == EINTR) {
goto Done;
}
result = POLL_ERROR; // epoll事件个数小于0，发生错误，直接跳转Done;
goto Done;
}
if (eventCount == 0) { //epoll事件个数等于0，发生超时，直接跳转Done;
result = POLL_TIMEOUT;
goto Done;
}
//循环遍历，处理所有的事件
for (int i = 0; i < eventCount; i++) {
int fd = eventItems[i].data.fd;
uint32_t epollEvents = eventItems[i].events;
if (fd == mWakeEventFd) {
if (epollEvents & EPOLLIN) {
awoken(); //已经唤醒了，则读取并清空管道数据
}
} else {
ssize_t requestIndex = mRequests.indexOfKey(fd);
if (requestIndex >= 0) {
int events = 0;
if (epollEvents & EPOLLIN) events |= EVENT_INPUT;
if (epollEvents & EPOLLOUT) events |= EVENT_OUTPUT;
if (epollEvents & EPOLLERR) events |= EVENT_ERROR;
if (epollEvents & EPOLLHUP) events |= EVENT_HANGUP;
//处理request，生成对应的reponse对象，push到响应数组
pushResponse(events, mRequests.valueAt(requestIndex));
}
}
}
Done: ;
//再处理Native的Message，调用相应回调方法
mNextMessageUptime = LLONG_MAX;
while (mMessageEnvelopes.size() != 0) {
nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0);
if (messageEnvelope.uptime <= now) {
{
sp<MessageHandler> handler = messageEnvelope.handler;
Message message = messageEnvelope.message;
mMessageEnvelopes.removeAt(0);
mSendingMessage = true;
mLock.unlock(); //释放锁
handler->handleMessage(message); // 处理消息事件
}
mLock.lock(); //请求锁
mSendingMessage = false;
result = POLL_CALLBACK; // 发生回调
} else {
mNextMessageUptime = messageEnvelope.uptime;
break;
}
}
mLock.unlock(); //释放锁
//处理带有Callback()方法的Response事件，执行Reponse相应的回调方法
for (size_t i = 0; i < mResponses.size(); i++) {
Response& response = mResponses.editItemAt(i);
if (response.request.ident == POLL_CALLBACK) {
int fd = response.request.fd;
int events = response.events;
void* data = response.request.data;
// 处理请求的回调方法
int callbackResult = response.request.callback->handleEvent(fd, events, data);
if (callbackResult == 0) {
removeFd(fd, response.request.seq); //移除fd
}
response.request.callback.clear(); //清除reponse引用的回调方法
result = POLL_CALLBACK; // 发生回调
}
}
return result;
}7.Looper::awoken()
void Looper::awoken() { uint64_t counter;
//不断读取管道数据，目的就是为了清空管道内容
TEMP_FAILURE_RETRY(read(mWakeEventFd, &counter, sizeof(uint64_t)));
}poll总结
pollInner()方法的处理流程：
先调用epoll_wait()，这是阻塞方法，用于等待事件发生或者超时，epoll_wait()主要在监听管道的read端是否有事件到来；
对于epoll_wait()返回，当且仅当以下3种情况出现：
POLL_ERROR，发生错误，直接跳转到Done；
POLL_TIMEOUT，发生超时，直接跳转到Done；
检测到管道有事件发生，则再根据情况做相应处理：
如果是管道读端产生事件，则直接读取管道的数据；
如果是其他事件，则处理request，生成对应的reponse对象，push到reponse数组；
进入Done标记位的代码段：
先处理Native的Message，调用Native 的Handler来处理该Message;
再处理Response数组，POLL_CALLBACK类型的事件；
从上面的流程，可以发现对于Request先收集，一并放入reponse数组，而不是马上执行。真正在Done开始执行的时候，是先处理native Message，再处理Request，说明native Message的优先级高于Request请求的优先级。
另外pollOnce()方法中，先处理Response数组中不带Callback的事件，再调用了pollInner()方法。
nativeWake()
nativeWake用于唤醒功能，什么时候应该唤醒和怎么唤醒呢？在添加消息到消息队列enqueueMessage(), 或者把消息从消息队列中全部移除quit()，再有需要时都会调用 nativeWake方法。
1.MessageQueue.enqueueMessage()
boolean enqueueMessage(Message msg, long when) {
... //将Message按时间顺序插入MessageQueue
if (needWake) {
nativeWake(mPtr);
}
}往消息队列添加Message时，需要根据mBlocked情况来决定是否需要调用nativeWake。我们可以看到mBlocked的赋值情况：
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;
}2.android_os_MessageQueue_nativeWake()
android_os_MessageQueue.cpp
static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {
NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
nativeMessageQueue->wake();
}3.NativeMessageQueue::wake()
android_os_MessageQueue.cpp
void NativeMessageQueue::wake() {
mLooper->wake();
}4.Looper::wake()
Looper.cpp
void Looper::wake() {
uint64_t inc = 1;
// 向管道mWakeEventFd写入字符1
ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t)));
if (nWrite != sizeof(uint64_t)) {
if (errno != EAGAIN) {
ALOGW("Could not write wake signal, errno=%d", errno);
}
}
}其中TEMP_FAILURE_RETRY 是一个宏定义， 当执行write失败后，会不断重复执行，直到执行成功为止，write的时候就会触发epoll的mWakeReadPipFdd唤醒进程，进而从MessageQueue的next方法，获取下一个msg。
sendMessage
接下来讲讲Native层如何向MessageQueue发送消息
1.sendMessage
void Looper::sendMessage(const sp<MessageHandler>& handler, const Message& message) { nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC); sendMessageAtTime(now, handler, message); }2.sendMessageDelayed
void Looper::sendMessageDelayed(nsecs_t uptimeDelay, const sp<MessageHandler>& handler,
const Message& message) { nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC); sendMessageAtTime(now + uptimeDelay, handler, message); }sendMessage(),sendMessageDelayed() 都是调用sendMessageAtTime()来完成消息插入。
3.sendMessageAtTime
void Looper::sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler>& handler,
const Message& message) {
size_t i = 0;
{ //请求锁
AutoMutex _l(mLock);
size_t messageCount = mMessageEnvelopes.size();
//找到message应该插入的位置i
while (i < messageCount && uptime >= mMessageEnvelopes.itemAt(i).uptime) {
i += 1;
}
MessageEnvelope messageEnvelope(uptime, handler, message);
mMessageEnvelopes.insertAt(messageEnvelope, i, 1);
//如果当前正在发送消息，那么不再调用wake()，直接返回。
if (mSendingMessage) {
return;
}
} //释放锁
//当把消息加入到消息队列的头部时，需要唤醒poll循环。
if (i == 0) {
wake();
}
}MessageQueue的native()方法，经过层层调用：
nativeInit()方法，最终实现由epoll机制中的epoll_create()/epoll_ctl()完成；
nativeDestroy()方法，最终实现由RefBase::decStrong()完成；
nativePollOnce()方法，最终实现由Looper::pollOnce()完成；
nativeWake()方法，最终实现由Looper::wake()调用write方法，向管道写入字符；
nativeIsPolling()，nativeSetFileDescriptorEvents()这两个方法类似，此处就不一一列举。
Native结构体和类
Looper.h/ Looper.cpp文件中，定义了Message结构体，消息处理类，回调类，Looper类。
Message结构体
struct Message {
Message() : what(0) { }
Message(int what) : what(what) { }
int what; // 消息类型
};消息处理类
class MessageHandler : public virtual RefBase {
protected:
virtual ~MessageHandler() { }
public:
virtual void handleMessage(const Message& message) = 0;
};WeakMessageHandler类，继承于MessageHandler类
class WeakMessageHandler : public MessageHandler {
protected:
virtual ~WeakMessageHandler();
public:
WeakMessageHandler(const wp<MessageHandler>& handler);
virtual void handleMessage(const Message& message);
private:
wp<MessageHandler> mHandler;
};
void WeakMessageHandler::handleMessage(const Message& message) {
sp<MessageHandler> handler = mHandler.promote();
if (handler != NULL) {
handler->handleMessage(message); //调用MessageHandler类的处理方法()
}
}回调类
LooperCallback类
class LooperCallback : public virtual RefBase {
protected:
virtual ~LooperCallback() { }
public:
//用于处理指定的文件描述符的poll事件
virtual int handleEvent(int fd, int events, void* data) = 0;
};SimpleLooperCallback类， 继承于LooperCallback类
class SimpleLooperCallback : public LooperCallback {
protected:
virtual ~SimpleLooperCallback();
public:
SimpleLooperCallback(Looper_callbackFunc callback);
virtual int handleEvent(int fd, int events, void* data);
private:
Looper_callbackFunc mCallback;
};
int SimpleLooperCallback::handleEvent(int fd, int events, void* data) {
return mCallback(fd, events, data); //调用回调方法
}Looper类
static const int EPOLL_SIZE_HINT = 8; //每个epoll实例默认的文件描述符个数
static const int EPOLL_MAX_EVENTS = 16; //轮询事件的文件描述符的个数上限其中Looper类的内部定义了Request，Response，MessageEnvelope这3个结构体，关系图如下：
struct Request { //请求结构体
int fd;
int ident;
int events;
int seq;
sp<LooperCallback> callback;
void* data;
void initEventItem(struct epoll_event* eventItem) const;
};
struct Response { //响应结构体
int events;
Request request;
};
struct MessageEnvelope { //信封结构体
MessageEnvelope() : uptime(0) { }
MessageEnvelope(nsecs_t uptime, const sp<MessageHandler> handler,
const Message& message) : uptime(uptime), handler(handler), message(message) {
}
nsecs_t uptime;
sp<MessageHandler> handler;
Message message;
};MessageEnvelope正如其名字，信封。MessageEnvelope里面记录着收信人(handler)，发信时间(uptime)，信件内容(message)
ALooper类
ALooper类定义在通过looper.cpp/looper.h（注意此文件是小写字母开头，与Looper.cpp不同）
static inline Looper* ALooper_to_Looper(ALooper* alooper) {
return reinterpret_cast<Looper*>(alooper);
}
static inline ALooper* Looper_to_ALooper(Looper* looper) {
return reinterpret_cast<ALooper*>(looper);
}ALooper类 与前面介绍的Looper类，更多的操作是通过ALooper_to_Looper(), Looper_to_ALooper()这两个方法转换完成的，也就是说ALooper类中定义的所有方法，都是通过转换为Looper类，再执行Looper中的方法。
红色虚线关系：Java层和Native层的MessageQueue通过JNI建立关联，彼此之间能相互调用，搞明白这个互调关系，也就搞明白了Java如何调用C++代码，C++代码又是如何调用Java代码。
蓝色虚线关系：Handler/Looper/Message这三大类Java层与Native层并没有任何的真正关联，只是分别在Java层和Native层的handler消息模型中具有相似的功能。都是彼此独立的，各自实现相应的逻辑。
WeakMessageHandler继承于MessageHandler类，NativeMessageQueue继承于MessageQueue类
另外，消息处理流程是先处理Native Message，再处理Native Request，最后处理Java Message。理解了该流程，也就明白有时上层消息很少，但响应时间却较长的真正原因。
