拨云见日---android异步消息机制源码分析(二)

在拨云见日---android异步消息机制源码分析(一)（http://my.oschina.net/u/1155515/blog/378460）中，我们了解了Java层异步消息机制的基本流程，可能细心的同学会发现java层中有很多native调用，其实java层仅仅是一个壳子，具体的实现全在native层，通过本篇文章，让我们继续抽丝剥茧，一步步揭开Android异步消息的面纱，分析不好之处，还请熟悉的童鞋指教

本文从实际使用角度出发，共分为2个部分

1、从子线程添加消息到消息线程时，从Java层到native的处理流程

2、消息线程消费消息的处理流程

一、从子线程添加消息到消息线程时，从Java层到native的处理流程

在子线程中，我们可以通过handler.sendMessage或Handler.postXXX（如post(Runable r)或postDelayed(Runnable r, long delayMillis)）向消息线程发送消息

而最终会调用到Java层MessageQueue.enqueueMessage把消息放入消息线程的MessageQueue（源码路径:android.os.MessageQueue）

booleanenqueueMessage(Messagemsg,longwhen){if(msg.isInUse()){thrownewAndroidRuntimeException(msg+"Thismessageisalreadyinuse.");}if(msg.target==null){thrownewAndroidRuntimeException("Messagemusthaveatarget.");}synchronized(this){if(mQuitting){RuntimeExceptione=newRuntimeException(msg.target+"sendingmessagetoaHandleronadeadthread");Log.w("MessageQueue",e.getMessage(),e);returnfalse;}msg.when=when;Messagep=mMessages;booleanneedWake;//判断入队消息是否需要立即处理if(p==null||when==0||when<p.when){//Newhead,wakeuptheeventqueueifblocked.msg.next=p;mMessages=msg;needWake=mBlocked;//如果消息循环线程处于阻塞中，则需要唤醒消息线程}else{//入队消息不需要立即处理，则根据消息处理时间插入到链表中合适位置.........Messageprev;for(;;){prev=p;p=p.next;if(p==null||when<p.when){break;}.............}msg.next=p;//invariant:p==prev.nextprev.next=msg;}//WecanassumemPtr!=0becausemQuittingisfalse.if(needWake){//唤醒nativeWake(mPtr);}}returntrue;}

通过上面代码，当入队消息需要立即处理并且消息线程处于阻塞时，会调用native函数nativeWake唤醒消息线程来处理消息

通过JNI定义，让我们看看nativeWake对应的native函数是什么

代码路径:frameworks\base\core\jni\android_os_MessageQueue.cpp

NativeMessageQueue::NativeMessageQueue():mPollEnv(NULL),mPollObj(NULL),mExceptionObj(NULL){mLooper=Looper::getForThread();if(mLooper==NULL){mLooper=newLooper(false);Looper::setForThread(mLooper);}}voidNativeMessageQueue::pollOnce(JNIEnv*env,jobjectpollObj,inttimeoutMillis){.......mLooper->pollOnce(timeoutMillis);.......}voidNativeMessageQueue::wake(){mLooper->wake();}staticvoidandroid_os_MessageQueue_nativePollOnce(JNIEnv*env,jobjectobj,jlongptr,jinttimeoutMillis){NativeMessageQueue*nativeMessageQueue=reinterpret_cast<NativeMessageQueue*>(ptr);nativeMessageQueue->pollOnce(env,obj,timeoutMillis);}staticvoidandroid_os_MessageQueue_nativeWake(JNIEnv*env,jclassclazz,jlongptr){NativeMessageQueue*nativeMessageQueue=reinterpret_cast<NativeMessageQueue*>(ptr);nativeMessageQueue->wake();}staticconstJNINativeMethodgMessageQueueMethods[]={/*name,signature,funcPtr*/{"nativeInit","()J",(void*)android_os_MessageQueue_nativeInit},{"nativeDestroy","(J)V",(void*)android_os_MessageQueue_nativeDestroy},{"nativePollOnce","(JI)V",(void*)android_os_MessageQueue_nativePollOnce},{"nativeWake","(J)V",(void*)android_os_MessageQueue_nativeWake},{"nativeIsPolling","(J)Z",(void*)android_os_MessageQueue_nativeIsPolling},{"nativeSetFileDescriptorEvents","(JII)V",(void*)android_os_MessageQueue_nativeSetFileDescriptorEvents},};intregister_android_os_MessageQueue(JNIEnv*env){intres=RegisterMethodsOrDie(env,"android/os/MessageQueue",gMessageQueueMethods,NELEM(gMessageQueueMethods));jclassclazz=FindClassOrDie(env,"android/os/MessageQueue");gMessageQueueClassInfo.mPtr=GetFieldIDOrDie(env,clazz,"mPtr","J");gMessageQueueClassInfo.dispatchEvents=GetMethodIDOrDie(env,clazz,"dispatchEvents","(II)I");returnres;}

调用关系链如下：

java层nativeWake->nativeMessageQueue.wake()->Looper.wake()

让我们通过源码一探Looper.wake()

代码路径:frameworks\native\jb-dev\libs\utils\Looper.cpp

voidLooper::wake(){#ifDEBUG_POLL_AND_WAKEALOGD("%p~wake",this);#endifssize_tnWrite;do{nWrite=write(mWakeWritePipeFd,"W",1);}while(nWrite==-1&&errno==EINTR);if(nWrite!=1){if(errno!=EAGAIN){ALOGW("Couldnotwritewakesignal,errno=%d",errno);}}}

mWakeWritePipeFd是个什么鬼？

为什么会向mWakeWritePipeFd写入一个字符？

别急，带着这些疑问我们继续往后面看，看了后面，前面的问题也就迎刃而解

现在我们简单的总结一下消息发送的流程：

1、在Java层通过Handler对象发送消息后，消息被放入消息线程的MessageQueue

2、消息入队后，会判断是否需要立即处理消息

3、如果需要立即处理消息且消息线程处于阻塞中，则唤醒消息线程

二、消息线程消费消息的处理流程

上面我们了解从子线程发送消息的流程，那么发送了消息后，消息线程是如何消费消息？

继续我们的脚步，让我们来一探究竟

通过前一篇文章（http://my.oschina.net/u/1155515/blog/378460）我们了解到消息线程通过调用Java层Looper.loop()进入消息循环，在消息循环中，又通过调用MessageQueue.next()不断的获取消息或者没有消息时阻塞

Messagenext(){intpendingIdleHandlerCount=-1;//-1onlyduringfirstiterationintnextPollTimeoutMillis=0;for(;;){...............//先调用native方法获阻塞到超时（超时时间由nextPollTimeoutMillis指定）或者被主动唤醒nativePollOnce(mPtr,nextPollTimeoutMillis);synchronized(this){//Trytoretrievethenextmessage.Returniffound.finallongnow=SystemClock.uptimeMillis();MessageprevMsg=null;Messagemsg=mMessages;...................//判断是有新消息到达还是超时if(msg!=null){//判断是否需要立即处理消息if(now<msg.when){//Nextmessageisnotready.Setatimeouttowakeupwhenitisready.nextPollTimeoutMillis=(int)Math.min(msg.when-now,Integer.MAX_VALUE);}else{//Gotamessage.////消息需要立即处理,则返回消息mBlocked=false;if(prevMsg!=null){prevMsg.next=msg.next;}else{mMessages=msg.next;}msg.next=null;if(false)Log.v("MessageQueue","Returningmessage:"+msg);msg.markInUse();returnmsg;}}....................//Processthequitmessagenowthatallpendingmessageshavebeenhandled.if(mQuitting){dispose();returnnull;}..................nextPollTimeoutMillis=0;}}

根据上面第一部分native层源码中JNI函数的定义，可以看到调用关系链如下：

java层nativePollOnce->nativeMessageQueue.pollOnce->Looper.pollOnce

intLooper::pollOnce(inttimeoutMillis,int*outFd,int*outEvents,void**outData){intresult=0;for(;;){...........if(result!=0){#ifDEBUG_POLL_AND_WAKEALOGD("%p~pollOnce-returningresult%d",this,result);#endifif(outFd!=NULL)*outFd=0;if(outEvents!=NULL)*outEvents=0;if(outData!=NULL)*outData=NULL;returnresult;}result=pollInner(timeoutMillis);}}

而Looper.pollOnce最终调用了Looper.pollInner

intLooper::pollInner(inttimeoutMillis){...............//Poll.intresult=ALOOPER_POLL_WAKE;structepoll_eventeventItems[EPOLL_MAX_EVENTS];//调用epoll_wait系统调用监听fd上事件，或者直到超时返回inteventCount=epoll_wait(mEpollFd,eventItems,EPOLL_MAX_EVENTS,timeoutMillis);.............//Checkforpollerror.if(eventCount<0){if(errno==EINTR){gotoDone;}ALOGW("Pollfailedwithanunexpectederror,errno=%d",errno);result=ALOOPER_POLL_ERROR;gotoDone;}//Checkforpolltimeout.if(eventCount==0){#ifDEBUG_POLL_AND_WAKEALOGD("%p~pollOnce-timeout",this);#endifresult=ALOOPER_POLL_TIMEOUT;gotoDone;}//Handleallevents.#ifDEBUG_POLL_AND_WAKEALOGD("%p~pollOnce-handlingeventsfrom%dfds",this,eventCount);#endiffor(inti=0;i<eventCount;i++){intfd=eventItems[i].data.fd;uint32_tepollEvents=eventItems[i].events;////判断是否是主动唤醒,if(fd==mWakeReadPipeFd){if(epollEvents&EPOLLIN){awoken();}else{ALOGW("Ignoringunexpectedepollevents0x%xonwakereadpipe.",epollEvents);}}.............}Done:;...................returnresult;}

这里又出现了mWakeReadPipeFd，让我们通过Looper构造函数看看mWakeReadPipeFd是个什么鬼

Looper::Looper(boolallowNonCallbacks):mAllowNonCallbacks(allowNonCallbacks),mSendingMessage(false),mResponseIndex(0),mNextMessageUptime(LLONG_MAX){intwakeFds[2];//创建管道intresult=pipe(wakeFds);LOG_ALWAYS_FATAL_IF(result!=0,"Couldnotcreatewakepipe.errno=%d",errno);mWakeReadPipeFd=wakeFds[0];mWakeWritePipeFd=wakeFds[1];result=fcntl(mWakeReadPipeFd,F_SETFL,O_NONBLOCK);LOG_ALWAYS_FATAL_IF(result!=0,"Couldnotmakewakereadpipenon-blocking.errno=%d",errno);result=fcntl(mWakeWritePipeFd,F_SETFL,O_NONBLOCK);LOG_ALWAYS_FATAL_IF(result!=0,"Couldnotmakewakewritepipenon-blocking.errno=%d",errno);//Allocatetheepollinstanceandregisterthewakepipe.mEpollFd=epoll_create(EPOLL_SIZE_HINT);LOG_ALWAYS_FATAL_IF(mEpollFd<0,"Couldnotcreateepollinstance.errno=%d",errno);//把读端管道添加到epoll监控列表并监听读端事件structepoll_eventeventItem;memset(&eventItem,0,sizeof(epoll_event));//zerooutunusedmembersofdatafieldunioneventItem.events=EPOLLIN;eventItem.data.fd=mWakeReadPipeFd;result=epoll_ctl(mEpollFd,EPOLL_CTL_ADD,mWakeReadPipeFd,&eventItem);LOG_ALWAYS_FATAL_IF(result!=0,"Couldnotaddwakereadpipetoepollinstance.errno=%d",errno);}

原来mWakeReadPipeFd只是管道的读端fd，可能童鞋们这时又有疑问

1、为什么要创建一个管道并监听读端事件？

2、为什么消息入队唤醒消息线程时，仅仅是向读端写一个字符？

通过上面的源码，我们知道当消息线程没有消息，则会一直阻塞到超时结束；但是若阻塞过程中，子线程发送一条消息，而这时消息线程还在阻塞中呢，那只能等消息线程阻塞结束才能处理消息，这样会造成消息处理延迟

可能聪明的童鞋会说，那我超时时间设置短一点行不行，这样看起来没问题，但是过短的超时时间基本上等于轮询，效率低不说还浪费CPU浪费电

所以常用的做法是：

1、创建一个pipe

2、pipe的读取端放入epoll监听队列

3、当需要立即唤醒消息线程时，子线程仅仅往读取端管道写一个字符就行

通过上面的分析，现在让我们总结一下

消息线程消费消息：

1、消息线程创建MessageQueue时，会在native层创建一个NativeMessageQueue

2、消息线程通过native调用进入阻塞状态，直到当有新消息到达被主动唤醒或者阻塞超时

3、消息线程被唤醒后，会判断是否有消息需要处理，如果有则返回消息处理，否则会继续步骤2直到退出

子线程传递消息流程：

1、MessageQueue中，消息按照执行时间从早到晚排列，当子线程发送消息后，根据消息执行时间判断是否需要立即唤醒消息线程

2、如果需要立即唤醒消息线程，则通过native端唤醒消息线程

3、消息线程被唤醒后，判断是否有消息需要处理

4、如果有新消息需要处理则返回，否则继续进入阻塞状态

一点思考：

Java层消息传递与消费为什么不用wait和Notify来实现，Native层仅仅是阻塞消息线程或者唤醒消息线程（Native层消息队列要强大得多，可以监听FD）有种杀鸡焉用牛刀的感觉

更多相关文章

随机推荐