AHandler AMessage ALooper消息机制

AHandler是Android native层实现的一个异步消息机制，在这个机制中所有的处理都是异步的，将变量封装到一个消息AMessage结构体中，然后放到队列中去，后台专门有一个线程ALooper会从这个队列中取出消息然后分发执行，执行函数就是AHandler实例的onMessageReceived。

1.主要类

主要的涉及的有以下三个类

• AMessage
  消息类，用于构造消息，通过post方法投递出去给ALooper
• ALooper
  轮询类，用于保存消息队列，然后将消息发送到对应的AHandler
• AHandler
  处理类，用于实际消息的处理

2.消息类型

发送的消息按照是否需要答复，分为两种

• 普通消息
• 需答复消息

3.模型简介

普通消息被Post到消息队列，Looper不断的从队列中取出消息并分发给相应的Handler处理。如果是需答复消息的话使用PostAndAwaitResponse，Handler处理完以后通过postReply将消息返回给发送方，发送方通过新建的responsemsg接收。

4.代码分析

4.1 初始化

伪代码

	sp<ALooper> looper = new ALooper;//创建一个ALooper实例sp<AHandlerReflector> mHandler = new AHandlerReflector();//创建一个AHandler实例looper->registerHandler(mHandler);//1. registerHandler会调用AHandler的setID方法将looper设置到Handler里面looper->start(true);//2. 根据参数创建并启动LooperThread

代码追踪

1. 注册mHandler 到looper

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

直接调用ALooperRoster的registerHandler，生成一个handler ID并保存起来

  ALooper::handler_id ALooperRoster::registerHandler(const sp<ALooper> &looper, const sp<AHandler> &handler) {Mutex::Autolock autoLock(mLock);if (handler->id() != 0) {CHECK(!"A handler must only be registered once.");  //一个handler在同一个looper中只能注册一次return INVALID_OPERATION;}HandlerInfo info;  //HandlerInfo 结构体，保存looper和handler，将其绑定在一起info.mLooper = looper;info.mHandler = handler;ALooper::handler_id handlerID = mNextHandlerID++;  //生成handler IDmHandlers.add(handlerID, info); //mHandlers为ALooperRoster的私有，保存handler列表handler->setID(handlerID, looper);return handlerID;
}

2. 启动looper循环

status_t ALooper::start(bool runOnCallingThread, bool canCallJava, int32_t priority) {if (runOnCallingThread) {{Mutex::Autolock autoLock(mLock);if (mThread != NULL || mRunningLocally) {return INVALID_OPERATION;}mRunningLocally = true;}do {} while (loop());return OK;}Mutex::Autolock autoLock(mLock);if (mThread != NULL || mRunningLocally) {return INVALID_OPERATION;}mThread = new LooperThread(this, canCallJava);  //开启一个线程status_t err = mThread->run(mName.empty() ? "ALooper" : mName.c_str(), priority);if (err != OK) {mThread.clear();}return err;
}

4.2 普通消息使用流程

实现一个AHandler的子类，主要是onMessageReceived方法
↓
创建一个ALooper对象
↓
创建消息AMessage进行投递Post
↓
onMessageReceived方法得到调用，处理实际的事务
伪代码
发送

sp<AMessage> msg = new AMessage(kWhatSaySomething, mHandler); //在AMessage的构造方法里获取Ahandler对应的Looper并保存
msg->post(); //1. 发送消息

处理

AHandlerReflector:: onMessageReceived //消息的具体的处理实现

代码追踪

1. 发送消息

status_t AMessage::post(int64_t delayUs) {sp<ALooper> looper = mLooper.promote();if (looper == NULL) {ALOGW("failed to post message as target looper for handler %d is gone.", mTarget);return -ENOENT;}looper->post(this, delayUs);  //进到looper线程return OK;
}

调用looper的post

void ALooper::post(const sp<AMessage> &msg, int64_t delayUs) {Mutex::Autolock autoLock(mLock);int64_t whenUs;if (delayUs > 0) {whenUs = GetNowUs() + delayUs;} else {whenUs = GetNowUs();}List<Event>::iterator it = mEventQueue.begin();while (it != mEventQueue.end() && (*it).mWhenUs <= whenUs) {++it;}Event event;event.mWhenUs = whenUs;event.mMessage = msg;if (it == mEventQueue.begin()) {mQueueChangedCondition.signal();  //如果之前消息队列为空，没有event，loop是处于阻塞状态的，此处当有消息进入时，先通知looperthread启动}mEventQueue.insert(it, event);  //进入消息队列
}

looper循环进行分发

bool ALooper::loop() {Event event;{Mutex::Autolock autoLock(mLock);if (mThread == NULL && !mRunningLocally) {return false;}if (mEventQueue.empty()) {mQueueChangedCondition.wait(mLock);  //如果消息队列为空，则等待，进入阻塞状态return true;}int64_t whenUs = (*mEventQueue.begin()).mWhenUs;int64_t nowUs = GetNowUs();if (whenUs > nowUs) {int64_t delayUs = whenUs - nowUs;mQueueChangedCondition.waitRelative(mLock, delayUs * 1000ll);return true;}event = *mEventQueue.begin();  //取出队列排头消息mEventQueue.erase(mEventQueue.begin());  //退出消息队列}event.mMessage->deliver();  //根据每个消息携带的handler做转发处理// NOTE: It's important to note that at this point our "ALooper" object// may no longer exist (its final reference may have gone away while// delivering the message). We have made sure, however, that loop()// won't be called again.return true;
}

返回工作线程

void AMessage::deliver() {sp<AHandler> handler = mHandler.promote();if (handler == NULL) {ALOGW("failed to deliver message as target handler %d is gone.", mTarget);return;}handler->deliverMessage(this);  //转发到相应的handler实例
}

转发到相应的Handler

void AHandler::deliverMessage(const sp<AMessage> &msg) {onMessageReceived(msg);  //Handler实例必须覆盖此方法，在这里做对应的处理，一般采用switch语句进行不同的处理mMessageCounter++;if (mVerboseStats) {uint32_t what = msg->what();ssize_t idx = mMessages.indexOfKey(what);if (idx < 0) {mMessages.add(what, 1);} else {mMessages.editValueAt(idx)++;}}
}

4.3 需要答复的消息使用流程

实现一个AHandler的子类，主要是onMessageReceived方法
↓
创建一个ALooper对象
↓
创建消息AMessage进行投递PostAndAwaitResponse
↓
onMessageReceived方法得到调用，处理实际的事务
↓
处理完消息后发送答复postReply
↓
发送端接收答复消息response
伪代码
发送/接收答复

sp<AMessage> msg = new AMessage(kWhatSaySomething, mHandler); //发送的消息
sp<AMessage> responsemsg; //接收存储答复消息
msg->postAndAwaitResponse(responsemsg);

处理/答复

AHandlerReflector:: onMessageReceived //消息的具体的处理实现
sp<AReplyToken> replyID;
CHECK(msg->senderAwaitsResponse(&replyID));
sp<AMessage> response = new AMessage;//创建答复消息
response->postReply(replyID);//发送答复消息

代码追踪

status_t AMessage::postAndAwaitResponse(sp<AMessage> *response) {sp<ALooper> looper = mLooper.promote();if (looper == NULL) {ALOGW("failed to post message as target looper for handler %d is gone.", mTarget);return -ENOENT;}sp<AReplyToken> token = looper->createReplyToken();  ////创建Token标识，此token用来唯一的标识这条信息if (token == NULL) {ALOGE("failed to create reply token");return -ENOMEM;}setObject("replyID", token);  //将此token添加到该Message的replyID字段looper->post(this, 0 /* delayUs */);  //发给looperreturn looper->awaitResponse(token, response);  //在此处就阻塞住，不返回了
}

post过程和普通消息的一致，先发送到looper，然后loop分发给对应的handler处理，不同的是此处在mRepliesCondition.wait(mRepliesLock)阻塞等待

status_t ALooper::awaitResponse(const sp<AReplyToken> &replyToken, sp<AMessage> *response) {// return status in case we want to handle an interrupted waitMutex::Autolock autoLock(mRepliesLock);CHECK(replyToken != NULL);while (!replyToken->retrieveReply(response)) {  //第一次会进入这个循环{Mutex::Autolock autoLock(mLock);if (mThread == NULL) {return -ENOENT;}}mRepliesCondition.wait(mRepliesLock);  //等待直到收到答复广播}return OK;
}

等收到mRepliesCondition.broadcast()以后，会执行retrieveReply，它的作用是将返回的reply拷贝到response

// if reply is not set, returns false; otherwise, it retrieves the reply and returns true
bool retrieveReply(sp<AMessage> *reply) {if (mReplied) {*reply = mReply;mReply.clear();}return mReplied;
}

所以上面的程序就在等待mRepliesCondition.broadcast()的发出，而这个broadcast是在哪里发出了，我们接着看
接着看消息的处理，和普通消息一致，在相应的handler实例的onMessageReceived中得到处理，处理完以后创建答复消息

sp<AReplyToken> replyID;
CHECK(msg->senderAwaitsResponse(&replyID));									
sp<AMessage> response = new AMessage;//创建答复消息
response->postReply(replyID);//发送答复消息

答复消息也是也Amessage的形式发出

status_t AMessage::postReply(const sp<AReplyToken> &replyToken) {if (replyToken == NULL) {ALOGW("failed to post reply to a NULL token");return -ENOENT;}sp<ALooper> looper = replyToken->getLooper();//根据replyToken获取消息所在的looper	if (looper == NULL) {ALOGW("failed to post reply as target looper is gone.");return -ENOENT;}return looper->postReply(replyToken, this); //转到looper
}

在looper这里首先将reply消息拷贝的位于AReplyToken的一个缓冲变量mReply，然后发出回复成果广播

status_t ALooper::postReply(const sp<AReplyToken> &replyToken, const sp<AMessage> &reply) {Mutex::Autolock autoLock(mRepliesLock);status_t err = replyToken->setReply(reply);  //将回复消息保存if (err == OK) {mRepliesCondition.broadcast();  /发出已回复广播}return err;
}

保存回复信息到缓存变量mReply

status_t AReplyToken::setReply(const sp<AMessage> &reply) {if (mReplied) {ALOGE("trying to post a duplicate reply");return -EBUSY;}CHECK(mReply == NULL);mReply = reply;  //保存回复信息mReplied = true;return OK;
}

[1]: https://blog.csdn.net/harman_zjc/article/details/53397945