Android Native层异步消息处理框架

*本文系作者工作学习总结，尚有不完善及理解不恰当之处，欢迎批评指正*

一、前言

　　在NuPlayer中，可以发现许多类似于下面的代码：

 1 //=======================================//
 2 NuPlayerDriver::NuPlayerDriver(pid_t pid)
 3     : ......
 4      mLooper(new ALooper) 
 5       ......{
 6     mLooper->setName("NuPlayerDriver Looper");
 7     mLooper->start(
 8             false, /* runOnCallingThread */
 9             true,  /* canCallJava */
10             PRIORITY_AUDIO);
11 
12     mPlayer = new NuPlayer(pid);
13     mLooper->registerHandler(mPlayer) 
14 } 
15 
16 //=======================================//
17 sp<AMessage> msg = new AMessage(kWhatPerformSeek, this);
18 msg->setInt32("generation", ++mSeekGeneration);
19 msg->setInt64("timeUs", seekTimeUs);
20 status_t err = msg->postAndAwaitResponse(&response);
21 
22 //=======================================//
23 void NuPlayer::RTSPSource::onMessageReceived(const sp<AMessage> &msg) {
24     switch (what) {
25         case XXX:
26         case YYY:
27         ......
28         default:
29     }
30 }

　　这就是android在native层实现的一个异步消息处理机制，在这个机制中所有的处理都是异步的。其基本的处理流程可概述如下：

将变量封装到一个消息AMessage结构体中，然后放到消息队列中去，后台专门有一个线程会从这个队列中取出消息并发送给指定的AHandler处理，在handler的onMessageReceived函数中根据AMessage的mWhat字段转向对应的分支进行处理。

　　在这里我们可能会产生这样的疑问：在很多类中都会有各种消息post出来，而后台的异步消息处理线程又是怎么知道将一个消息发送给哪个类的onMessageReceived函数处理呢？

　　要搞明白这个问题，就需要分析android在native层的异步消息处理框架。

二、基础类分析

　　在android native层的异步消息处理框架中涉及到的类主要有：ALooper、AHandler、AMessage、ALooperRoster、LooperThread等。

　　在android开源代码中，这些类的声明头文件位于：/frameworks/av/include/media/stagefright/foundation/ ，实现功能的源代码位于：/frameworks/av/media/libstagefright/foundation/ ，下面就分别对这些类进行分析。

　　1、AMessage -- 消息的载体

　　结构体AMessage是消息的载体，在传递消息的过程中可以携带各种信息。

 1 //AMessage类简析
 2 struct AMessage : public RefBase {
 3     //构造函数，在其中会对AMessage的两个重要数据成员mWhat和mTarget进行初始化设置。
 4     AMessage();
 5     AMessage(uint32_t what, const sp<const AHandler> &handler);
 6     void setWhat(uint32_t what);//设置mWhat
 7     uint32_t what() const;//返回mWhat
 8     void setTarget(const sp<const AHandler> &handler);//设置mTarget
 9     //一系列的set和find函数，用于在传递消息的过程中携带各种信息
10     void setInt32(const char *name, int32_t value);
11     void setInt64(const char *name, int64_t value);
12     //....
13     bool findInt32(const char *name, int32_t *value) const;
14     bool findInt64(const char *name, int64_t *value) const;
15     //投递消息到消息队列中
16     status_t post(int64_t delayUs = 0);
17     // Posts the message to its target and waits for a response (or error) before returning.
18     status_t postAndAwaitResponse(sp<AMessage> *response);
19 protected:
20     virtual ~AMessage();//析构函数
21 private:
22     friend struct ALooper; // deliver()
23     //两个重要的私有数据成员：
24     //mWhat指明这是一个什么消息,用于在onMessageReceived处理分支中进行匹配。
25     //mTarget用于后台线程在处理这个消息时判断发送给哪一个类处理。
26     uint32_t mWhat;
27     ALooper::handler_id mTarget;
28 
29     wp<AHandler> mHandler;
30     wp<ALooper> mLooper;

　　进一步分析AMessage对象的构造函数，首先我们查看AMessage的source code，如下：

 1 AMessage::AMessage(uint32_t what, const sp<const AHandler> &handler)
 2     : mWhat(what),
 3       mNumItems(0) {
 4     setTarget(handler);
 5 }
 6 
 7 void AMessage::setTarget(const sp<const AHandler> &handler) {
 8     if (handler == NULL) {
 9         mTarget = 0;
10         mHandler.clear();
11         mLooper.clear();
12     } else {
13         mTarget = handler->id();
14         mHandler = handler->getHandler();
15         mLooper = handler->getLooper();
16     }
17 }

　　在构造函数中设置mWhat的值为指定的what值，用于指明这是一个什么消息，以便在onMessageReceived函数中找到匹配的分支进行处理。然后会调用setTarget函数，来设置mTarget为指定handler的mID值，mHandler为指定的handler，mLooper为与handler相关的ALooper，这样后台处理线程就可以据此判断将该消息发送给哪一个类(handler)进行处理了。

　　构造消息并投递出去的过程，如下示例：

1 void NuPlayer::start() {
2     (new AMessage(kWhatStart, this))->post();
3 }

　　2、ALooper类

　　这个类定义了消息循环和后台处理线程。

 1 //ALooper类简析
 2 struct ALooper : public RefBase {
 3     //定义的两个整型变量类型
 4     typedef int32_t event_id;//event id
 5     typedef int32_t handler_id;//handler id
 6     
 7     ALooper();//构造函数
 8     
 9     // Takes effect in a subsequent call to start().
10     void setName(const char *name); 
11 
12     handler_id registerHandler(const sp<AHandler> &handler);//注册handler
13     void unregisterHandler(handler_id handlerID);//注销handler
14     //启动后台处理线程处理事件
15     status_t start(
16             bool runOnCallingThread = false,
17             bool canCallJava = false,
18             int32_t priority = PRIORITY_DEFAULT
19             );
20 protected:
21     virtual ~ALooper();    
22 private:
23     friend struct AMessage;       // post()
24     //事件的结构体封装
25     struct Event {  
26         int64_t mWhenUs;
27         sp<AMessage> mMessage;
28     };
29     AString mName;
30     List<Event> mEventQueue;
31     struct LooperThread;
32     sp<LooperThread> mThread;//后台处理线程,LooperThread继承自Thread
33     // posts a message on this looper with the given timeout
34     void post(const sp<AMessage> &msg, int64_t delayUs);
35     bool loop();
36 }

　　3、LooperThread -- 后台处理线程

 1 struct ALooper::LooperThread : public Thread {
 2     LooperThread(ALooper *looper, bool canCallJava)
 3         : Thread(canCallJava),
 4           mLooper(looper),
 5           mThreadId(NULL) {
 6     }
 7 ....
 8     virtual bool threadLoop() {
 9         return mLooper->loop();
10     }
11 ....
12 protected:
13     virtual ~LooperThread() {}
14 
15 private:
16     ALooper *mLooper;
17     android_thread_id_t mThreadId;
18 };

　　LooperThread类继承自Thread，其函数threadLoop()用于在线程启动后进行消息处理。在调用run()函数后便会执行threadLoop()函数中的程序，当其返回值为true时，该函数会被再次调用。

　　4、AHandler -- 消息处理类的父类

　　AHandler类用于对接收到的消息进行对应的处理。下面我们结合source code进行分析如下：

 1 struct AHandler : public RefBase {
 2     AHandler()    //构造函数
 3         : mID(0),    //初始化mID为0
 4           mVerboseStats(false),
 5           mMessageCounter(0) {
 6     }
 7 
 8     ALooper::handler_id id() const {//这个函数用于返回内部变量mID的值，其初始值为0，但是可以通过setID()设置
 9         return mID;
10     }
11 
12 
13 protected:
14     virtual void onMessageReceived(const sp<AMessage> &msg) = 0;
15 
16 private:
17     friend struct AMessage;      // deliverMessage()
18     friend struct ALooperRoster; // setID()
19 
20     ALooper::handler_id mID;
21     wp<ALooper> mLooper;
22     //setID()在其友元类ALooperRoster的registerHandler()函数中调用。
23     inline void setID(ALooper::handler_id id, wp<ALooper> looper) {
24         mID = id;
25         mLooper = looper;
26     }
27 
28 };

　　5、ALooperRoster类

　　ALooperRoster可以看做是ALooper的一个辅助类，主要用于完成消息handler的注册与注销工作。下面结合source code分析如下：

 1 //ALooperRoster简析
 2 struct ALooperRoster {
 3     ALooperRoster();
 4 
 5     ALooper::handler_id registerHandler(//注册handler
 6             const sp<ALooper> looper, const sp<AHandler> &handler);
 7 
 8     void unregisterHandler(ALooper::handler_id handlerID);//注销handler
 9     void unregisterStaleHandlers();//注销旧有的handler,在构造新的ALooper对象时会调用
10 private:
11     struct HandlerInfo {
12         wp<ALooper> mLooper;
13         wp<AHandler> mHandler;
14     };
15 
16     Mutex mLock;
17     KeyedVector<ALooper::handler_id, HandlerInfo> mHandlers;
18     ALooper::handler_id mNextHandlerID;
19 };

三、异步消息处理框架

　　首先我们先copy一段代码来看一看异步消息处理框架的搭建过程：

1 sp<ALooper> mLooper;
2 mLooper(new ALooper)//构造函数的初始化列表中
3 mLooper->setName("NuPlayerDriver Looper");
4 mLooper->start(false, true, PRIORITY_AUDIO);
5 mPlayer = new NuPlayer;// struct NuPlayer : public AHandler
6 mLooper->registerHandler(mPlayer);

　　搭建框架1

　　调用ALooper的构造函数，实例化一个新的ALooper对象：

1 ALooper::ALooper()
2     : mRunningLocally(false) {
3     // clean up stale AHandlers. Doing it here instead of in the destructor avoids
4     // the side effect of objects being deleted from the unregister function recursively.
5     gLooperRoster.unregisterStaleHandlers();
6 }

　　搭建框架2

　　调用ALooper::setName(const char *name)设置ALooper的名字：

1 void ALooper::setName(const char *name) {
2     mName = name;
3 }

　　搭建框架3

　　调用ALooper::start()启动异步消息处理后台线程：

 1 status_t ALooper::start(
 2         bool runOnCallingThread, bool canCallJava, int32_t priority) {
 3     // ......
 4     
 5     if (mThread != NULL || mRunningLocally) {
 6         return INVALID_OPERATION;
 7     }
 8     mThread = new LooperThread(this, canCallJava);//构造LooperThread对象
 9     status_t err = mThread->run( //run()之后就会执行LooperThread::threadLooper()函数，在threadlooper()函数中又会调用ALooper::loop()函数
10             mName.empty() ? "ALooper" : mName.c_str(), priority);
11     if (err != OK) {
12         mThread.clear();
13     }
14     return err;
15 }

　　搭建框架4

　　调用ALooper::registerHandler()注册消息处理器类AHandler:

1 ALooper::handler_id ALooper::registerHandler(const sp<AHandler> &handler) {
2     return gLooperRoster.registerHandler(this, handler);
3 }

　　从这个函数的定义可以看出，能作为handler进行注册的类都必须继承自AHandler这个类，注册的过程也是交给了LooperRoster::registerHandler()函数来处理的。接下来我们分析LooperRoster::registerHandler()处理的具体过程。

 1 //注册handler的过程可以理解为建立ALooper对象与AHandler对象间的联系
 2 ALooper::handler_id ALooperRoster::registerHandler(
 3         const sp<ALooper> looper, const sp<AHandler> &handler) {
 4     Mutex::Autolock autoLock(mLock);
 5     //由前面对AHandler的介绍可知，在AHandler构造函数中，其mID初始值为0.
 6     //调用AHandler::id()可获取mID值
 7     //调用AHandler::setID()可设置mID值
 8     if (handler->id() != 0) {
 9         //任何一个AHandler对象只能注册一次，未注册的mID==0
10         CHECK(!"A handler must only be registered once.");
11         return INVALID_OPERATION;
12     }
13     //结构体ALooperRoster::HandlerInfo的定义为：
14     //struct HandlerInfo {
15     //    wp<ALooper> mLooper;
16     //    wp<AHandler> mHandler;
17     //};
18     HandlerInfo info;
19     info.mLooper = looper;
20     info.mHandler = handler;
21     //mNextHandlerID是ALooperRoster的私有数据成员，其初始值为1，每注册一个handler其值加1
22     //gALooperRoster是一个ALooperRoster类型的全局对象，利用ALooperRoster的registerHandler()
23     //函数注册handler过程中mNextHandlerID从1开始递增，这样保证每一个AHandler对象均有一个独一无二的handler_id
24     //最后在发送Message进行处理时就是根据这个独一无二的handler_id找到对应得handler进行处理的。
25     ALooper::handler_id handlerID = mNextHandlerID++;
26     //在gALooperRoster中有定义KeyedVector<ALooper::handler_id, HandlerInfo> mHandlers;
27     //mHandlers是一个存储handler_id与HandlerInfo对的容器
28     //注：一个looper可以有多个handler，但一个handler只能有一个对应的looper
29     mHandlers.add(handlerID, info);
30     //设置handler的mID
31     handler->setID(handlerID, looper);
32 
33     return handlerID;
34 }

　　注：LooperRoster对象gLooperRoster并不是某一个ALooper对象的私有数据成员，而是一个全局的对象，这样就保证了在整个程序运行中利用gLooperRoster::registerHandler()进行handler注册时，每一个AHandler对象均会有一个独一无二的handler_id.

四、消息处理

　　消息处理首先要产生消息并将其投递出去，然后等待处理。下面是一段产生消息并投递的实例：

1     sp<AMessage> msg = new AMessage(kWhatPerformSeek, mReflector->id());
2     msg->setInt32("generation", ++mSeekGeneration);
3     msg->setInt64("timeUs", seekTimeUs);
4     msg->post(200000ll);

　　可以看到通过调用AMessage::post()函数将消息投递出去，在分析post()函数前，我们先了解一下AMessage对象如何建立与ALooper、AHandler的对象的联系？

　　首先我们看一下AMessage的构造函数：

1 AMessage::AMessage(uint32_t what, const sp<const AHandler> &handler)
2 　　: mWhat(what),
3 　　mNumItems(0) {
4 　　　　setTarget(handler);
5 }

　　在构造函数中传递进来what(这是个什么消息)和handler(由谁处理)两个参数，首先初始化mWhat字段，然后调用setTarget函数:

 1 void AMessage::setTarget(const sp<const AHandler> &handler) {
 2     if (handler == NULL) {
 3         mTarget = 0;
 4         mHandler.clear();
 5         mLooper.clear();
 6     } else {
 7         mTarget = handler->id();
 8         mHandler = handler->getHandler();
 9         mLooper = handler->getLooper();
10     }
11 }

　　在setTarget函数中将mTarget初始化为handler_id,mTarget用于后台线程在处理这个消息时判断发送给哪一个类处理。同时获取与handler相关联的ALooper对象--mLooper = handler->getLooper()。这样就建立了AMessage对象与ALooper、AHandler的对象的联系。

　　接下来我们分析AMessage::post()函数，source code如下：

 1 status_t AMessage::post(int64_t delayUs) {
 2     sp<ALooper> looper = mLooper.promote();
 3     if (looper == NULL) {
 4         ALOGW("failed to post message as target looper for handler %d is gone.", mTarget);
 5         return -ENOENT;
 6     }
 7 
 8     looper->post(this, delayUs);
 9     return OK;
10 }

可以看到，post之后调用了ALooper::post()来处理，下面是器source code：

 1 void ALooper::post(const sp<AMessage> &msg, int64_t delayUs) {
 2     Mutex::Autolock autoLock(mLock);
 3 
 4     int64_t whenUs;
 5     if (delayUs > 0) {
 6         whenUs = GetNowUs() + delayUs;
 7     } else {
 8         whenUs = GetNowUs();
 9     }
10 
11     List<Event>::iterator it = mEventQueue.begin();
12     while (it != mEventQueue.end() && (*it).mWhenUs <= whenUs) {
13         ++it;
14     }
15 //上面这一段代码，主要是遍历消息队列中的消息的时间，然后和我们的消息做对比，最终目的就是所有的消息必须按照时间先后顺序放在队列中等待执行。
16 //构造一个Event消息，将我们的msg设置进去
17     Event event;
18     event.mWhenUs = whenUs;
19     event.mMessage = msg;
20 
21     if (it == mEventQueue.begin()) {
22         mQueueChangedCondition.signal();
23     }
24 //将我们的消息放入到消息队列中，等待执行
25     mEventQueue.insert(it, event);
26 }

五、后台线程消息处理

　　前面有提到创建ALooper对象后调用ALooper::start()函数启动了一个后台消息处理线程，接下来我们进行详细分析：

 1 status_t ALooper::start(
 2         bool runOnCallingThread, bool canCallJava, int32_t priority) {
 3     //......
 4     if (mThread != NULL || mRunningLocally) {
 5         return INVALID_OPERATION;
 6     }
 7     mThread = new LooperThread(this, canCallJava);//创建LooperThread对象
 8     //**********************************************************************
 9     status_t err = mThread->run(       //
10             mName.empty() ? "ALooper" : mName.c_str(), priority);
11     //**********************************************************************        
12     if (err != OK) {
13         mThread.clear();
14     }
15     return err;
16 }

　　调用LooperThread::run()函数启动后台处理线程，就会转去执行LooperThread::threadLoop()函数：

virtual bool threadLoop() {
    return mLooper->loop();
}

　　在threadLoop函数中又会调用创建LooperThread对象时传进来的ALooper的loop()函数：

 1 bool ALooper::loop() {
 2     Event event;
 3     
 4     {
 5         Mutex::Autolock autoLock(mLock);
 6         ....
 7         //取出Event队列的队头消息，取出后然后把它从队列中删掉
 8         event = *mEventQueue.begin();
 9         mEventQueue.erase(mEventQueue.begin());
10     }
11     //调用AMessage的deliver函数发送出去执行
12     event.mMessage->deliver();
13 
14     return true;
15 }

　　注：如果ALooper::loop()返回true，则threadLoop()函数会被再次调用，后台线程继续执行，如果返回false则后台线程停止运行。

　在ALooper::loop()函数中取出队头的消息并调用AMessage::deliver()函数发送出去：

1 void AMessage::deliver() {
2     sp<AHandler> handler = mHandler.promote();
3     if (handler == NULL) {
4         ALOGW("failed to deliver message as target handler %d is gone.", mTarget);
5         return;
6     }
7     handler->deliverMessage(this);
8 }

　　AMessage::deliver()函数中，

　　首先得到处理message的handler -- sp<AHandler> handler = mHandler.promote();

　　然后会调用AHandler::deliverMessage()函数 -- handler->deliverMessage(this);

　　转到处理该message的handler进行处理：

 1 void AHandler::deliverMessage(const sp<AMessage> &msg) {
 2     onMessageReceived(msg);
 3     mMessageCounter++;
 4 
 5     if (mVerboseStats) {
 6         uint32_t what = msg->what();
 7         ssize_t idx = mMessages.indexOfKey(what);
 8         if (idx < 0) {
 9             mMessages.add(what, 1);
10         } else {
11             mMessages.editValueAt(idx)++;
12         }
13     }
14 }

　　在AHandler::deliverMessage()函数中调用了AHandler::onMessageReceived(msg)函数，上面有提到该函数中多多个case分支，根据message的what字段转到对应的分支进行处理，比如在NuPlayer中：

 1 void NuPlayer::onMessageReceived(const sp<AMessage> &msg) {
 2     switch (msg->what()) {
 3         case kWhatSetDataSource:
 4         case kWhatVideoNotify:
 5         case kWhatAudioNotify:
 6         case kWhatRendererNotify:
 7         case kWhatMoreDataQueued:
 8         case kWhatReset:
 9         case kWhatSeek:
10         case kWhatPause:
11         case kWhatResume:
12         default:
13     }
14 }

　　至此消息处理的整个流程结束。

六、总结

　　1. 首先需要构造一个AMessage对象，必须携带两个参数：what(什么消息)和handler(谁处理)；
　　2. ALooper中有一个后台线程--LooperThread，该线程维护着一个消息队列List<Event> mEventQueue，线程函数不断从这个队列中取出消息执行；
　　3. 消息的发送和取出需要调用ALooper，AMessage，AHandler等的函数。

微信扫一扫，关注玖零日记，获取更多相关资讯及源码 -- 虽无面朝大海，依旧春暖花开