android多媒体框架之流媒体AHandler消息机制----base on jellybean(十) .

为什么我们要谈论流媒体的消息机制呢？因为在流媒体中，类似于我们写APP的时候，为了不阻塞UI线程，我们把利用handler，把UI线程分开异步执行，使用handler去执行某项比较费时的操作，然后异步更新UI线程。流媒体中也是类似的，因为联网，codec都很费时，需要异步执行。handler是java的实现机制，而我们下面要讲的AHandler就是基于C++的实现了。

我们知道handler消息机制，构成就必须包括一个Loop，message。那么对应的AHandler，也应该有对应的ALooper， AMessage。下面我们将以实例化NUplayerDrriver和setdataSource为例来具体讲述AHandler消息机制。

首先看下NuplayerDriver的构造函数，这是流媒体初始化函数。

static sp<MediaPlayerBase> createPlayer(player_type playerType, void* cookie,
notify_callback_f notifyFunc){
caseNU_PLAYER:
ALOGV(" createNuPlayer");
p = newNuPlayerDriver;
}

NuPlayerDriver::NuPlayerDriver()
: mResetInProgress(false),
mPrepareInProgress(false),
mIsPrepared(false),
mDurationUs(-1),
mPositionUs(-1),
mNumFramesTotal(0),
mNumFramesDropped(0),
mLooper(new ALooper),-----创建一个新的ALooper
mState(UNINITIALIZED),
mAtEOS(false),
mStartupSeekTimeUs(-1) {
mLooper->setName("NuPlayerDriverLooper");----给该Looper取名字，以便与AHandler一一对应
mLooper->start(
false, /* runOnCallingThread */
true, /* canCallJava */
PRIORITY_AUDIO);-------------启动该Looper

mPlayer = new NuPlayer;------------创建一个AHandler即Nuplayer
mLooper->registerHandler(mPlayer);-----把该AHandler注册到Looper中，具体的实现我们往后看
mPlayer->setDriver(this);
}

看看ALooper的启动函数：

status_t ALooper::start(
bool runOnCallingThread, boolcanCallJava, int32_t priority) {
if (runOnCallingThread)
{------runOnCallingThread开始为false，不走这里
…………
}

Mutex::Autolock autoLock(mLock);

if (mThread != NULL || mRunningLocally) {
return INVALID_OPERATION;
}

mThread = newLooperThread(this, canCallJava);----新建一个thread

status_t err =mThread->run(
mName.empty() ?"ALooper" : mName.c_str(), priority);----looper线程启动
if (err != OK) {
mThread.clear();
}

return err;
}
看下关键步骤注册Handler：

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

ALooper::handler_idALooperRoster::registerHandler(
const sp<ALooper> looper, constsp<AHandler> &handler) {
Mutex::Autolock autoLock(mLock);

if (handler->id() != 0) {
CHECK(!"A handler must only beregistered once.");
return INVALID_OPERATION;
}
HandlerInfo info;
info.mLooper = looper;-----
“NuPlayerDriver Looper”
info.mHandler = handler;------nuplayer
ALooper::handler_idhandlerID = mNextHandlerID++;
mHandlers.add(handlerID, info);-------KeyedVector<ALooper::handler_id,HandlerInfo> mHandlers;
handler->setID(handlerID);------设置handlerID，以便发送message时找到对应的handler
return handlerID;
}
ALooperRoster::ALooperRoster()
: mNextHandlerID(1),------------------从1开始
mNextReplyID(1) {
}

有了LOOPER，也有了对应的handler，看看如何发送消息给LOOPER，交个相应的handler去处理。我们以setdataSource方法为例：

Nuplayer本身也是个AHandler，因为其继承自AHandler。

structNuPlayer : public AHandler {
}
我们看看其父类AHandler：
struct AHandler : public RefBase {
AHandler()
: mID(0){
}
ALooper::handler_id id() const {
return mID;
}
sp<ALooper> looper();
protected:
virtual voidonMessageReceived(const sp<AMessage> &msg) = 0;---处理消息函数
private:
friend struct ALooperRoster;
ALooper::handler_id mID;
void setID(ALooper::handler_id id) {
mID = id;
}
DISALLOW_EVIL_CONSTRUCTORS(AHandler);
};
以setdataSource为例看看如何传递message

void NuPlayer::setDataSource(
const char *url, constKeyedVector<String8, String8> *headers) {
（1） sp<AMessage> msg =new AMessage(kWhatSetDataSource, id());
size_t len = strlen(url);
………..
elseif ((!strncasecmp(url, "http://", 7) || !strncasecmp(url,"https://", 8))
&& ((len >= 4&& !strcasecmp(".sdp", &url[len - 4]))
|| strstr(url,".sdp?"))) {
source = newRTSPSource(url, headers, mUIDValid, mUID, true);
mSourceType = kRtspSource;
}
……….
（2）msg->setObject("source", source);
（3）msg->post();
}

首先新建一个AMessage的实例，传入的参数为事件的名称以及处理该消息的Handlerid，该id在 mLooper->registerHandler(mPlayer);方法中设置上。

我们看下AMessage：

AMessage::AMessage(uint32_twhat, ALooper::handler_id target)
: mWhat(what),
mTarget(target),
mNumItems(0) {
}

void AMessage::setObject(const char *name, const sp<RefBase> &obj) {
setObjectInternal(name, obj, kTypeObject);
}
void AMessage::setObjectInternal(
const char *name, constsp<RefBase> &obj, Type type) {
Item *item = allocateItem(name);
item->mType = type;

if (obj != NULL) { obj->incStrong(this);}
item->u.refValue = obj.get();
}

POST 过程:

void AMessage::post(int64_t delayUs) {
gLooperRoster.postMessage(this, delayUs);----调用ALooperRoster的postMessage函数
}

status_tALooperRoster::postMessage(
const sp<AMessage> &msg,int64_t delayUs) {
Mutex::Autolock autoLock(mLock);
return postMessage_l(msg, delayUs);
}

status_t ALooperRoster::postMessage_l(
const sp<AMessage> &msg,int64_t delayUs) {
ssize_t index =mHandlers.indexOfKey(msg->target());--target即为Handler_id

if (index < 0) {
ALOGW("failed to post message.Target handler not registered.");
return -ENOENT;
}

const HandlerInfo &info =mHandlers.valueAt(index);---根据handler_id找到HandlerInfo

sp<ALooper>looper = info.mLooper.promote();----根据我们注册的HandlerInfo找到相应的ALooper，我们现在就是“NuPlayerDriver
Looper”

if (looper == NULL) {
ALOGW("failed to post message."
"Target handler %d stillregistered, but object gone.",
msg->target());

mHandlers.removeItemsAt(index);
return -ENOENT;
}

looper->post(msg,delayUs);---往“NuPlayerDriver Looper”里传递消息

return OK;
}

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();
}

mEventQueue.insert(it,event);----往消息队列里插入消息
}

当队列里有消息时便会触发loop函数：

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());
}

gLooperRoster.deliverMessage(event.mMessage);
return true;
}

void ALooperRoster::deliverMessage(const sp<AMessage> &msg) {
sp<AHandler> handler;
{
Mutex::Autolock autoLock(mLock);
ssize_t index = mHandlers.indexOfKey(msg->target());
if (index < 0) {
ALOGW("failed to delivermessage. Target handler not registered.");
return;
}
const HandlerInfo &info =mHandlers.valueAt(index);
handler =info.mHandler.promote();

if (handler == NULL) {
ALOGW("failed to delivermessage. "
"Target handler %dregistered, but object gone.",
msg->target());
mHandlers.removeItemsAt(index);
return;
}
}
handler->onMessageReceived(msg);------对应为Nuplayer
}
void NuPlayer::onMessageReceived(const sp<AMessage> &msg) {
switch (msg->what()) {
case kWhatSetDataSource:
{
………………………………………
mSource = static_cast<Source*>(obj.get());

sp<AMessage> notify = newAMessage(kWhatSourceNotify, id());
mSource->setNotify(notify);
mSource->connect();-------------RTSPSource

break;
}
}
至此我们的Ahandler的流程讲完了，大致就是启动一个threadLooper，监听looper的消息队列是否有变化，如有交个相应的Handler去处理。