Android中的消息處理機制大量依賴于Handler。每個Handler都有對應的Looper，用于不斷地從對應的MessageQueue中取出消息處理。
一直以來，覺得MessageQueue應該是Java層的抽象，然而事實上MessageQueue的主要部分在Native層中。
自己對MessageQueue在Native層的工作不太熟悉，借此機會分析一下。
一、MessageQueue的創建
當需要使用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為線程本地存儲區；每個線程僅有一個Looper sThreadLocal.set(new Looper(quitAllowed));}private Looper(boolean quitAllowed) { //創建出MessageQueue mQueue = new MessageQueue(quitAllowed); mThread = Thread.currentThread();}1 NativeMessageQueue
我們看看MessageQueue的構造函數：
MessageQueue(boolean quitAllowed) { mQuitAllowed = quitAllowed; //mPtr的類型為long？ mPtr = nativeInit();}MessageQueue的構造函數中就調用了native函數，我們看看android_os_MessageQueue.cpp中的實現：
static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) { //MessageQueue的Native層實體 NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue(); ............ //這里應該類似與將指針轉化成long類型，放在Java層保存；估計Java層使用時，會在native層將long變成指針，就可以操作隊列了 return reinterpret_cast<jlong>(nativeMessageQueue);}我們跟進NativeMessageQueue的構造函數：
NativeMessageQueue::NativeMessageQueue() : mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) { //創建一個Native層的Looper，也是線程唯一的 mLooper = Looper::getForThread(); if (mLooper == NULL) { mLooper = new Looper(false); Looper::setForThread(mLooper); }}從代碼來看，Native層和Java層均有Looper對象，應該都是操作MessageQueue的。MessageQueue在Java層和Native層有各自的存儲結構，分別存儲Java層和Native層的消息。
2 Native層的looper
我們看看Native層looper的構造函數：
Looper::Looper(bool allowNonCallbacks) : mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false), mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false), mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) { //此處創建了個fd mWakeEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); ....... rebuildEpollLocked();}在native層中，MessageQueue中的Looper初始化時，還調用了rebuildEpollLocked函數，我們跟進一下：
void Looper::rebuildEpollLocked() { // Close old epoll instance if we have one. if (mEpollFd >= 0) { close(mEpollFd); } // Allocate the new epoll instance and register the wake pipe. mEpollFd = epoll_create(EPOLL_SIZE_HINT); ............ struct epoll_event eventItem; memset(& eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union eventItem.events = EPOLLIN; eventItem.data.fd = mWakeEventFd; //在mEpollFd上監聽mWakeEventFd上是否有數據到來 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對應fd上數據的到來 int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, & eventItem); ............ }}從native層的looper來看，我們知道Native層依賴于epoll來驅動事件處理。此處我們先保留一下大致的映像，后文詳細分析。
二、使用MessageQueue
1 寫入消息
Android中既可以在Java層向MessageQueue寫入消息，也可以在Native層向MessageQueue寫入消息。我們分別看一下對應的操作流程。
1.1 Java層寫入消息
Java層向MessageQueue寫入消息，依賴于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) { ..... 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; //在頭部插入數據，如果之前MessageQueue是阻塞的，那么現在需要喚醒 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; } //不是第一個異步消息時，needWake置為false 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按執行時間插入到原有的隊列中，然后根據情況調用nativeAwake函數。
我們跟進一下nativeAwake：
void NativeMessageQueue::wake() { mLooper->wake();}void Looper::wake() { uint64_t inc = 1; //就是向mWakeEventFd寫入數據 ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t))); .............}在native層的looper初始化時，我們提到過native層的looper將利用epoll來驅動事件，其中構造出的epoll句柄就監聽了mWakeEventFd。
實際上從MessageQueue中取出數據時，若沒有數據到來，就會利用epoll進行等待；因此當Java層寫入消息時，將會將喚醒處于等待狀態的MessageQueue。
在后文介紹從MessageQueue中提取消息時，將再次分析這個問題。
1.2 Native層寫入消息
Native層寫入消息，依賴于Native層looper的sendMessage函數：
void Looper::sendMessage(const sp<MessageHandler>& handler, const Message& message) { nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC); sendMessageAtTime(now, handler, message);}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(); while (i < messageCount && uptime >= mMessageEnvelopes.itemAt(i).uptime) { i += 1; } //將message包裝成一個MessageEnvelope對象 MessageEnvelope messageEnvelope(uptime, handler, message); mMessageEnvelopes.insertAt(messageEnvelope, i, 1); // Optimization: If the Looper is currently sending a message, then we can skip // the call to wake() because the next thing the Looper will do after processing // messages is to decide when the next wakeup time should be. In fact, it does // not even matter whether this code is running on the Looper thread. if (mSendingMessage) { return; } } // Wake the poll loop only when we enqueue a new message at the head. if (i == 0) { //若插入在隊列頭部，同樣利用wake函數觸發epoll喚醒 wake(); }}以上就是向MessageQueue中加入消息的主要流程，接下來我們看看從MessageQueue中取出消息的流程。
2、提取消息
當Java層的Looper對象調用loop函數時，就開始使用MessageQueue提取消息了：
public static void loop() { final Looper me = myLooper(); ....... for (;;) { Message msg = queue.next(); // might block ....... try { //調用Message的處理函數進行處理 msg.target.dispatchMessage(msg); }........ }}此處我們看看MessageQueue的next函數：
Message next() { //mPtr保存了NativeMessageQueue的指針 final long ptr = mPtr; ....... int pendingIdleHandlerCount = -1; // -1 only during first iteration int nextPollTimeoutMillis = 0; for (;;) { if (nextPollTimeoutMillis != 0) { //會調用Native函數，最終調用IPCThread的talkWithDriver，將數據寫入Binder驅動或者讀取一次數據 //不知道在此處進行這個操作的理由？ Binder.flushPendingCommands(); } //處理native層的數據，此處會利用epoll進行blocked nativePollOnce(ptr, nextPollTimeoutMillis); synchronized (this) { 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; //完成next記錄的存儲 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; } //MessageQueue中引入了IdleHandler接口，即當MessageQueue沒有數據處理時，調用IdleHandler進行一些工作 //pendingIdleHandlerCount表示待處理的IdleHandler，初始為-1 if (pendingIdleHandlerCount < 0 && (mMessages == null || now < mMessages.when)) { //mIdleHandlers的size默認為0，調用接口addIdleHandler才能增加 pendingIdleHandlerCount = mIdleHandlers.size(); } if (pendingIdleHandlerCount <= 0) { // No idle handlers to run. Loop and wait some more. mBlocked = true; continue; } //將待處理的IdleHandler加入到PendingIdleHandlers中 if (mPendingIdleHandlers == null) { mPendingIdleHandlers = new IdleHandler[Math.max(pendingIdleHandlerCount, 4)]; } //調用ArrayList.toArray(T[])節省每次分配的開銷；畢竟對于Message.Next這樣調用頻率較高的函數，能省一點就是一點 mPendingIdleHandlers = mIdleHandlers.toArray(mPendingIdleHandlers); } 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 { //執行實現類的queueIdle函數，返回值決定是否繼續保留 keep = idler.queueIdle(); } catch (Throwable t) { Log.wtf(TAG, "IdleHandler threw exception", t); } if (!keep) { synchronized (this) { mIdleHandlers.remove(idler); } } } pendingIdleHandlerCount = 0; nextPollTimeoutMillis = 0; }}
整個提取消息的過程，大致上如上圖所示。
可以看到在Java層，Looper除了要取出MessageQueue的消息外，還會在隊列空閑期執行IdleHandler定義的函數。
2.1 nativePollOnce
現在唯一的疑點是nativePollOnce是如何處理Native層數據的，我們看看對應的native函數：
static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj, jlong ptr, jint timeoutMillis) { //果然Java層調用native層MessageQueue時，將long類型的ptr變為指針 NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr); nativeMessageQueue->pollOnce(env, obj, timeoutMillis);}void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) { mPollEnv = env; mPollObj = pollObj; //最后還是進入到Native層looper的pollOnce函數 mLooper->pollOnce(timeoutMillis); mPollObj = NULL; mPollEnv = NULL; if (mExceptionObj) { ......... }}看看native層looper的pollOnce函數：
//timeoutMillis為超時等待時間。值為-1時，表示無限等待直到有事件到來；值為0時，表示無需等待//outFd此時為null，含義是：存儲產生事件的文件句柄//outEvents此時為null，含義是：存儲outFd上發生了哪些事件，包括可讀、可寫、錯誤和中斷//outData此時為null，含義是：存儲上下文數據，其實調用時傳入的參數int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) { int result = 0; for (;;) { //處理response，目前我們先不關注response的內含 while (mResponseIndex < mResponses.size()) { const Response& response = mResponses.itemAt(mResponseIndex++); int ident = response.request.ident; if (ident >= 0) { 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; } } //根據pollInner的結果，進行操作 if (result != 0) { if (outFd != NULL) *outFd = 0; if (outEvents != NULL) *outEvents = 0; if (outData != NULL) *outData = NULL; return result; } //主力還是靠pollInner result = pollInner(timeoutMillis); }}跟進一下pollInner函數：
int Looper::pollInner(int timeoutMillis) { // Adjust the timeout based on when the next message is due. //timeoutMillis是Java層事件等待事件 //native層維持了native message的等待時間 //此處其實就是選擇最小的等待時間 if (timeoutMillis != 0 && mNextMessageUptime != LLONG_MAX) { nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC); int messageTimeoutMillis = toMillisecondTimeoutDelay(now, mNextMessageUptime); if (messageTimeoutMillis >= 0 && (timeoutMillis < 0 || messageTimeoutMillis < timeoutMillis)) { timeoutMillis = messageTimeoutMillis; } } int result = POLL_WAKE; //pollInner初始就清空response mResponses.clear(); mResponseIndex = 0; // We are about to idle. mPolling = true; //利用epoll等待mEpollFd監控的句柄上事件到達 struct epoll_event eventItems[EPOLL_MAX_EVENTS]; int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis); // No longer idling. mPolling = false; // Acquire lock. mLock.lock(); //重新調用rebuildEpollLocked時，將使得epoll句柄能夠監聽新加入request對應的fd if (mEpollRebuildRequired) { mEpollRebuildRequired = false; rebuildEpollLocked(); goto Done; } // Check for poll error. if (eventCount < 0) { if (errno == EINTR) { goto Done; } ...... result = POLL_ERROR; goto Done; } // Check for poll timeout. if (eventCount == 0) { result = POLL_TIMEOUT; goto Done; } for (int i = 0; i < eventCount; i++) { if (fd == mWakeEventFd) { if (epollEvents & EPOLLIN) { //前面已經分析過，當java層或native層有數據寫入隊列時，將寫mWakeEventFd，以觸發epoll喚醒 //awoken將讀取并清空mWakeEventFd上的數據 awoken(); } else { ......... } } else { //epoll同樣監聽的request對應的fd 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; //存儲這個fd對應的response pushResponse(events, mRequests.valueAt(requestIndex)); } else { .......... } } }Done：
// Invoke pending message callbacks. mNextMessageUptime = LLONG_MAX; //處理Native層的Message while (mMessageEnvelopes.size() != 0) { nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC); const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0); if (messageEnvelope.uptime <= now) { // Remove the envelope from the list. // We keep a strong reference to the handler until the call to handleMessage // finishes. Then we drop it so that the handler can be deleted *before* // we reacquire our lock. { sp<MessageHandler> handler = messageEnvelope.handler; Message message = messageEnvelope.message; mMessageEnvelopes.removeAt(0); mSendingMessage = true; mLock.unlock(); //處理Native Message handler->handleMessage(message); } mLock.lock(); mSendingMessage = false; result = POLL_CALLBACK; } else { // The last message left at the head of the queue determines the next wakeup time. mNextMessageUptime = messageEnvelope.uptime; break; } } // Release lock. mLock.unlock(); //處理帶回調函數的response 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; //調用response的callback int callbackResult = response.request.callback->handleEvent(fd, events, data); if (callbackResult == 0) { removeFd(fd, response.request.seq); } response.request.callback.clear(); result = POLL_CALLBACK; } } return result;}
說實話native層的代碼寫的很亂，該函數的功能比較多。
如上圖所示，在nativePollOnce中利用epoll監聽是否有數據到來，然后處理native message、native response。
最后，我們看看如何在native層中加入request。
3 添加監控請求
native層增加request依賴于looper的接口addFd：
//fd表示需要監聽的句柄//ident的含義還沒有搞明白//events表示需要監聽的事件，例如EVENT_INPUT、EVENT_OUTPUT、EVENT_ERROR和EVENT_HANGUP中的一個或多個//callback為事件發生后的回調函數//data為回調函數對應的參數int Looper::addFd(int fd, int ident, int events, Looper_callbackFunc callback, void* data) { return addFd(fd, ident, events, callback ? new SimpleLooperCallback(callback) : NULL, data);}結合上文native層輪詢隊列的操作，我們大致可以知道：addFd的目的，就是讓native層的looper監控新加入的fd上是否有指定事件發生。
如果發生了指定的事件，就利用回調函數及參數構造對應的response。
native層的looper處理response時，就可以執行對應的回調函數了。
看看實際的代碼：
int Looper::addFd(int fd, int ident, int events, const sp<LooperCallback>& callback, void* data) { ........ { AutoMutex _l(mLock); //利用參數構造一個request Request request; request.fd = fd; request.ident = ident; request.events = events; request.seq = mNextRequestSeq++; request.callback = callback; request.data = data; if (mNextRequestSeq == -1) mNextRequestSeq = 0; // reserve sequence number -1 struct epoll_event eventItem; request.initEventItem(&eventItem); //判斷之前是否已經利用該fd構造過Request ssize_t requestIndex = mRequests.indexOfKey(fd); if (requestIndex < 0) { //mEpollFd新增一個需監聽fd int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, & eventItem); ....... mRequests.add(fd, request); } else { //mEpollFd修改舊的fd對應的監聽事件 int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_MOD, fd, & eventItem); if (epollResult < 0) { if (errno == ENOENT) { // Tolerate ENOENT because it means that an older file descriptor was // closed before its callback was unregistered and meanwhile a new // file descriptor with the same number has been created and is now // being registered for the first time. epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, & eventItem); ....... } //發生錯誤重新加入時，安排EpollRebuildLocked，將讓epollFd重新添加一次待監聽的fd scheduleEpollRebuildLocked(); } mRequests.replaceValueAt(requestIndex, request); } }}對加入監控請求的處理，在上文介紹pollInner函數時已做分析，此處不再贅述。
三、總結
1、流程總結
MessageQueue的整個流程包括了Java部分和Native部分，從圖中可以看出Native層的比重還是很大的。我們結合上圖回憶一下整個MessageQueue對應的處理流程：
1、Java層創建Looper對象時，將會創建Java層的MessageQueue；Java層的MessageQueue初始化時，將利用Native函數創建出Native層的MessageQueue。
2、Native層的MessageQueue初始化后，將創建對應的Native Looper對象。Native對象初始化時，將創建對應epollFd和WakeEventFd。其中，epollFd將作為epoll的監聽句柄，初始時epollFd僅監聽WakeEventFd。
3、圖中紅色線條為Looper從MessageQueue中取消息時，處理邏輯的流向。
3.1、當Java層的Looper開始循環時，首先需要通過JNI函數調用Native Looper進行pollOnce的操作。
3.2、Native Looper開始運行后，需要等待epollFd被喚醒。當epollFd等待超時或監聽的句柄有事件到來，Native Looper就可以開始處理事件了。
3.3、在Native層，Native Looper將先處理Native MessageQueue中的消息，再調用Response對應的回調函數。
3.4、本次循環中，Native層事件處理完畢后，才開始處理Java層中MessageQueue的消息。若MessageQueue中沒有消息需要處理，并且MessageQueue中存在IdleHandler時，將調用IdleHandler定義的處理函數。
圖中藍色部分為對應的函數調用：
在Java層：
利用MessageQueue的addIdleHandler，可以為MessageQueue增加IdleHandler；
利用MessageQueue的enqueueMessage，可以向MessageQueue增加消息；必要時將利用Native函數向Native層的WakeEventFd寫入消息，以喚醒epollFd。
在Native層：
利用looper:sendMessage，可以為Native MessageQueue增加消息；同樣，要時將向Native層的WakeEventFd寫入消息，以喚醒epollFd；
利用looper:addFd，可以向Native Looper注冊監聽請求，監聽請求包含需監聽的fd、監聽的事件及對應的回調函數等，監聽請求對應的fd將被成為epollFd監聽的對象。當被監聽的fd發生對應的事件后，將會喚醒epollFd，此時將生成對應response加入的response List中，等待處理。一旦response被處理，就會調用對應的回調函數。
2、注意事項
MessageQueue在Java層和Native層有各自的存儲結構，可以分別增加消息。從處理邏輯來看，會優先處理native層的Message，然后處理Native層生成的response，最后才是處理Java層的Message。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。
新聞熱點
疑難解答
