編輯:關於Android編程
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。
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持本站。
本文開發一個基於Service的音樂播放器,音樂由後台運行的Service負責播放,當後台的播放狀態發生變化時,程序將會通過發送廣播通知前台Activity更新界面;當點
相信大家都已經可以熟練使用ListView和GridView,大神們估計都在使用RecyclerView了。如果還在使用ListView,你肯定有這樣的一個深刻的感受,那
先上一張效果圖吧:說說這個東西的來源吧,今天突然想到笛卡爾心形圖,想去看看能不能畫個心出來,但是看到一篇不錯的文章,那篇文章羅列了很多關於心形的函數方程,這可把我高興壞了
上一篇介紹了Animation動畫其一:Tween補間動畫。 這篇文章接下來介紹Animation另一種動畫形式:Frame逐幀動畫。 Frame動畫是一系列圖片按照一定