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去处理。