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