Android NuPlayer要点详解

1、Android NuPlayer要点详解1、AHandler机制首先介绍NuPlayer中无处不在的AHandler机制 frameworks/av/include/media/stagefright/foundation/ frameworks/av/media/libstagefright/foundation/ AHandler是Android native层实现的一个异步消息机制，在这个机制中所有的处理都是异步的，将变量封装到一个消息AMessage结构体中，然后放到队列中去，后台专门有一个线程会从这个队列中取出消息然后执行，执行函数就是onMessageReceived。Ahandl

2、er机制包括以下几个类AMessage消息类，用于构造消息，通过post方法投递出去给ALooperstatus_t AMessage:post(int64_t delayUs) sp<ALooper> looper = mLmote(); if (looper = NULL) ALOGW("failed to post message as target looper for handler %d is gone.", mTarget); return -ENOENT; looper->post(this, delayUs); retu

3、rn OK;void AMessage:deliver() sp<AHandler> handler = mHmote(); if (handler = NULL) ALOGW("failed to deliver message as target handler %d is gone.", mTarget); return; handler->deliverMessage(this); /see AHandlerdeliverMessage，前面通过looper post最后就是调用这里的deliever送到handler手里AHa

4、ndler 消息处理类，一般当做父类，继承该类的子类需要实现onMessageReceived方法void AHandler:deliverMessage(const sp<AMessage> &msg) onMessageReceived(msg); mMessageCounter+;.ALooper 与Ahander一一对应，负责存储消息并分发Ahandler的消息，与AMessage一对多关系/ posts a message on this looper with the given timeoutvoid ALooper:post(const sp<AMes

5、sage> &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

6、; event.mWhenUs = whenUs; event.mMessage = msg; if (it = mEventQueue.begin() mQueueChangedCondition.signal(); mEventQueue.insert(it, event);-status_t ALooper:start( bool runOnCallingThread, bool canCallJava, int32_t priority) if (runOnCallingThread) Mutex:Autolock autoLock(mLock); if (mThread != NUL

7、L | 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() ? "AL

8、ooper" : mName.c_str(), priority); if (err != OK) mThread.clear(); return err;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 =

9、(*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();/see AHandl

10、er.deliverMessage. return true;LooperThread 此线程调用ALooper的loop方法来分发消息 virtual status_t readyToRun() mThreadId = androidGetThreadId(); return Thread:readyToRun(); virtual bool threadLoop() return mLooper->loop();ALooperRoaster 与Handler是一对多的关系， 管理Looper和Handler一一对应关系，负责释放stale handlerALooper:handler

11、_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."); return INVALID_OPERATION; HandlerInfo info; info.mLooper = looper; info.mH

12、andler = handler; ALooper:handler_id handlerID = mNextHandlerID+;/一对一 mHandlers.add(handlerID, info);/一对多 handler->setID(handlerID, looper); return handlerID;void ALooperRoster:unregisterHandler(ALooper:handler_id handlerID) Mutex:Autolock autoLock(mLock); ssize_t index = mHandlers.indexOfKey(han

13、dlerID); if (index < 0) return; const HandlerInfo &info = mHandlers.valueAt(index); sp<AHandler> handler = info.mHmote(); if (handler != NULL) handler->setID(0, NULL); mHandlers.removeItemsAt(index);void ALooperRoster:unregisterStaleHandlers() Vector<sp<ALooper> &g

14、t; activeLoopers; Mutex:Autolock autoLock(mLock); for (size_t i = mHandlers.size(); i > 0;) i-; const HandlerInfo &info = mHandlers.valueAt(i); sp<ALooper> looper = info.mLmote(); if (looper = NULL) ALOGV("Unregistering stale handler %d", mHandlers.keyAt(i); mHandlers

15、.removeItemsAt(i); else / At this point 'looper' might be the only sp<> keeping / the object alive. To prevent it from going out of scope / and having ALooper call this method again recursively / and then deadlocking because of the Autolock above, add / it to a Vector which will go out

16、 of scope after the lock / has been released. activeLoopers.add(looper); 异步消息机制的创建sp<ALooper> mLooper = new ALooper; /创建一个Alooper实例sp<AHandlerReflector> mHandler = new AHandlerReflector /创建一个Ahandler实例mLooper->setName(“xxxxx”); /设置looper名字mLooper->start(false, true, PRIORITY_XXX);

17、/根据参数创建并启动 looper threadmLooper->regiserHandler(mHandler); /register handler 会调用AHandler的setID方法将looper设置到Handler里去Post消息sp<AMessage> msg = new AMessage(kWhatSayGoodbye, mHandler); /在AMessage的构造方法里获取Ahandler对应的Looper并保存msg->post(); / 调用looper的post方法Message Post的调用过程Message:postALooper:po

18、stmEventQueue.insertmQueueChangedCondition.signal() /如果之前没有event，通知looper threadALooper:loop()if (mEventQueue.empty() /如果消息队列为空，则等待mQueueChangedCondition.wait(mLock);return true;event = *mEventQueue.begin();event.mMessage->deliver();AHandler:deliverMessageAHandlerReflector: onMessageReceived具体的实现

19、NuPlayer下面就进入我们的正题，NuPlayer frameworks/av/media/libmediaplayerservice/nuplayer/NuPlayerDriver NuPlayerDriver是对NuPlayer的封装，继承MediaPlayerInterface接口。通过NuPlayer来实现播放的功能。看这部分代码的方法就是先看NuPlayerDriver里面干了啥，转头就去找NuPlayer里面的实现，一般都要再去NuPlayer的onMessageReceive中看消息的响应，最后回到NuPlayerDriver的各种notify中看流程的周转，下面附上一张播放

20、器状态机流转图NuPlayerDriver:NuPlayerDriver(pid_t pid) : mState(STATE_IDLE), /对应播放器状态机的初始化状态 mIsAsyncPrepare(false), mAsyncResult(UNKNOWN_ERROR), mSetSurfaceInProgress(false), mDurationUs(-1), mPositionUs(-1), mSeekInProgress(false), mLooper(new ALooper), mPlayerFlags(0), mAtEOS(false), mLooping(false), mA

21、utoLoop(false) ALOGV("NuPlayerDriver(%p)", this); /和前面所述的异步消息创建机制相符mLooper->setName("NuPlayerDriver Looper"); mLooper->start( false, /* runOnCallingThread */ true, /* canCallJava */ PRIORITY_AUDIO);/mPlayer即NuPlayer，继承于AHandler mPlayer = AVNuFactory:get()->createNuPlayer

22、(pid); mLooper->registerHandler(mPlayer); mPlayer->setDriver(this);NuPlayerDriver:NuPlayerDriver() ALOGV("NuPlayerDriver(%p)", this); mLooper->stop(); /整个NuPlayerDriver就是一个大ALooperAVNuFactory 负责关键组件的create，通过它能看到： 1.每一个NuPlayer对应一个进程 2.数据流从Source-Decoder-Renderer，中间由AMessages驱动sp&

23、lt;NuPlayer> AVNuFactory:createNuPlayer(pid_t pid) return new NuPlayer(pid);sp<NuPlayer:DecoderBase> AVNuFactory:createPassThruDecoder( const sp<AMessage> &notify, const sp<NuPlayer:Source> &source, const sp<NuPlayer:Renderer> &renderer) return new NuPlayer:Dec

24、oderPassThrough(notify, source, renderer);sp<NuPlayer:DecoderBase> AVNuFactory:createDecoder( const sp<AMessage> &notify, const sp<NuPlayer:Source> &source, pid_t pid, const sp<NuPlayer:Renderer> &renderer) return new NuPlayer:Decoder(notify, source, pid, renderer

25、);sp<NuPlayer:Renderer> AVNuFactory:createRenderer( const sp<MediaPlayerBase:AudioSink> &sink, const sp<AMessage> &notify, uint32_t flags) return new NuPlayer:Renderer(sink, notify, flags);下面分别分析Source, Decoder, RendererSource以setDataSource为切入点status_t NuPlayerDriver:setDat

26、aSource(const sp<IStreamSource> &source) ALOGV("setDataSource(%p) stream source", this); Mutex:Autolock autoLock(mLock); if (mState != STATE_IDLE) return INVALID_OPERATION; mState = STATE_SET_DATASOURCE_PENDING; mPlayer->setDataSourceAsync(source);/因为driver只是NuPlayer的封装，所以还是要去

27、调用NuPlayer完成实际动作 while (mState = STATE_SET_DATASOURCE_PENDING) mCondition.wait(mLock); return mAsyncResult;void NuPlayer:setDataSourceAsync(const sp<IStreamSource> &source) sp<AMessage> msg = new AMessage(kWhatSetDataSource, this); sp<AMessage> notify = new AMessage(kWhatSource

28、Notify, this); msg->setObject("source", new StreamingSource(notify, source); msg->post();/到了NuPlayer中，也不是直接进行操作，而是先发个消息，验证前面所说的一切都由AMessage驱动void NuPlayer:onMessageReceived(const sp<AMessage> &msg) switch (msg->what() case kWhatSetDataSource:/实际的处理在这里 ALOGV("kWhatSe

29、tDataSource"); CHECK(mSource = NULL); status_t err = OK; sp<RefBase> obj; CHECK(msg->findObject("source", &obj); if (obj != NULL) Mutex:Autolock autoLock(mSourceLock); mSource = static_cast<Source *>(obj.get();/赋值给mSource else err = UNKNOWN_ERROR; CHECK(mDriver != N

30、ULL); sp<NuPlayerDriver> driver = mDmote(); if (driver != NULL) driver->notifySetDataSourceCompleted(err);/通知driver设置完毕 break; .void NuPlayerDriver:notifySetDataSourceCompleted(status_t err) Mutex:Autolock autoLock(mLock); CHECK_EQ(mState, STATE_SET_DATASOURCE_PENDING); mAsyncResul

31、t = err; mState = (err = OK) ? STATE_UNPREPARED : STATE_IDLE;/回到driver中，流转播放器状态进入下一阶段 mCondition.broadcast();下面就来看看具体有哪些source，它们都继承自NuPlayer:Source（NuPlayerSource.h & NuPlayerSource.cpp） 1.HTTP-进一步判断是以下的哪一种：HTTPLiveSource,RTSPSource,GenericSource 2.File-GenericSource 3.StreamSource-StreamingSou

32、rce 4.DataSource-GenericSourceGenericSourcenuplayer/GenericSource.h & GenericSource.cpp几个水位static int64_t kLowWaterMarkUs = 2000000ll; / 2secsstatic int64_t kHighWaterMarkUs = 5000000ll; / 5secsstatic int64_t kHighWaterMarkRebufferUs = 15000000ll; / 15secs,这一个是新增加的水位static const ssize_t kLowWate

33、rMarkBytes = 40000;static const ssize_t kHighWaterMarkBytes = 200000;status_t NuPlayer:GenericSource:initFromDataSource() init extractor;get track info and metadatavoid NuPlayer:GenericSource:prepareAsync() if (mLooper = NULL) mLooper = new ALooper; mLooper->setName("generic"); mLooper-

34、>start(); mLooper->registerHandler(this); sp<AMessage> msg = new AMessage(kWhatPrepareAsync, this); msg->post();status_t NuPlayer:GenericSource:feedMoreTSData() return OK;LiveSessionlibstagefright/httplive/LiveSession.h & cpp/ static/ Bandwidth Switch Mark Defaultsconst int64_t Li

35、veSession:kUpSwitchMarkUs = 15000000ll;const int64_t LiveSession:kDownSwitchMarkUs = 20000000ll;const int64_t LiveSession:kUpSwitchMarginUs = 5000000ll;const int64_t LiveSession:kResumeThresholdUs = 100000ll;/ Buffer Prepare/Ready/Underflow Marksconst int64_t LiveSession:kReadyMarkUs = 5000000ll;con

36、st int64_t LiveSession:kPrepareMarkUs = 1500000ll;const int64_t LiveSession:kUnderflowMarkUs = 1000000ll;与Fetcher,Bandwidth Estimater(和ExoPlayer一样是滑动窗口平均),switching,Buffering相关的操作都在这里HTTPLiveSourcenuplayer目录下enum Flags / Don't log any URLs.不在log中记录URL kFlagIncognito = 1, ;NuPlayer:HTTPLiveSource

37、:HTTPLiveSource( if (headers) /也搞了一个header机制 mExtraHeaders = *headers; ssize_t index = mExtraHeaders.indexOfKey(String8("x-hide-urls-from-log"); if (index >= 0) mFlags |= kFlagIncognito; mExtraHeaders.removeItemsAt(index); void NuPlayer:HTTPLiveSource:prepareAsync() if (mLiveLooper = NU

38、LL) mLiveLooper = new ALooper;/一如既往的ALooper mLiveLooper->setName("http live"); mLiveLooper->start(); mLiveLooper->registerHandler(this); sp<AMessage> notify = new AMessage(kWhatSessionNotify, this); mLiveSession = new LiveSession( notify, (mFlags & kFlagIncognito) ? Live

39、Session:kFlagIncognito : 0, mHTTPService); mLiveLooper->registerHandler(mLiveSession); mLiveSession->connectAsync(/HTTPLiveSource包含LiveSession，很多实际的工作都由LiveSession完成 mURL.c_str(), mExtraHeaders.isEmpty() ? NULL : &mExtraHeaders);ATSParser frameworks/av/media/libstagefright/mpeg2ts/ATSParse

40、r.cpp 就是一个TS Parser，虽然也叫Axx，但是没有消息机制在里面StreamingSource nuplayer目录void NuPlayer:StreamingSource:prepareAsync() if (mLooper = NULL) mLooper = new ALooper; mLooper->setName("streaming"); mLooper->start();/何其相似 mLooper->registerHandler(this); notifyVideoSizeChanged(); notifyFlagsChang

41、ed(0); notifyPrepared();StreamingSource中的数据由onReadBuffer驱动，最后的EOS,Discontiunity等都交给ATSParser去处理，ATSParser又最终交给AnotherPacketSource去做真正的处理实际上，这里提到的三个Source最后都会用到AnotherPacketSourcevoid NuPlayer:StreamingSource:onReadBuffer() for (int32_t i = 0; i < kNumListenerQueuePackets; +i) char buffer188; sp&l

42、t;AMessage> extra; ssize_t n = mStreamListener->read(buffer, sizeof(buffer), &extra);/实际用NuPlayerStreamListener完成工作 if (n = 0) ALOGI("input data EOS reached."); mTSParser->signalEOS(ERROR_END_OF_STREAM);/EOS了 setError(ERROR_END_OF_STREAM); break; else if (n = INFO_DISCONTINUIT

43、Y) int32_t type = ATSParser:DISCONTINUITY_TIME; int32_t mask; if (extra != NULL && extra->findInt32( IStreamListener:kKeyDiscontinuityMask, &mask) if (mask = 0) ALOGE("Client specified an illegal discontinuity type."); setError(ERROR_UNSUPPORTED); break; type = mask; mTSPars

44、er->signalDiscontinuity( (ATSParser:DiscontinuityType)type, extra); else if (n < 0) break; else if (buffer0 = 0x00) / XXX legacy if (extra = NULL) extra = new AMessage; uint8_t type = buffer1; if (type & 2) int64_t mediaTimeUs; memcpy(&mediaTimeUs, &buffer2, sizeof(mediaTimeUs); ex

45、tra->setInt64(IStreamListener:kKeyMediaTimeUs, mediaTimeUs); mTSParser->signalDiscontinuity( (type & 1) = 0) ? ATSParser:DISCONTINUITY_TIME : ATSParser:DISCONTINUITY_FORMATCHANGE, extra); else status_t err = mTSParser->feedTSPacket(buffer, sizeof(buffer); if (err != OK) ALOGE("TS P

46、arser returned error %d", err); mTSParser->signalEOS(err); setError(err); break; AnotherPacketSourceframeworks/av/media/libstagefright/mpeg2ts 可以类比ExoPlayer中的chunk source，同时负责buffer管理，EOSDiscontinuity的处理等等前面三个Source最后都会落到AnotherPacketSourcebool AnotherPacketSource:hasBufferAvailable(status_t

47、 *finalResult) Mutex:Autolock autoLock(mLock); *finalResult = OK; if (!mEnabled) return false; if (!mBuffers.empty() /一个ABuffer List，其实就是一个环形缓冲 return true; *finalResult = mEOSResult; return false;void AnotherPacketSource:queueDiscontinuity( ATSParser:DiscontinuityType type, const sp<AMessage> &extra, bool discard) Mutex:Autolock autoLock(mLock); if (discard) / Leave only discontinuities in the queue. mEOSResult = OK; mLastQueuedTimeUs = 0; mLatestEnqueuedMeta = NULL; if (type = ATSParser:DISCONTINUITY_NONE) return; mDiscontinuitySe