前言

這里所謂的拉流從就是指從本地文件或者遠(yuǎn)程文件不停獲取壓縮的音視頻數(shù)據(jù)包并緩存在本地待解碼的過程，用一張圖形象的畫出來其過程如下：

image.png

• 拉流模塊

這里要有個拉流線程讓拉流模塊在此線程中不停的工作，它需要滿足忙時工作閑時休眠等待蕉斜，對于拉流模塊心赶，在ffmpeg的世界里也可以通俗的稱為解析器典唇，不同的協(xié)議從其中獲取數(shù)據(jù)的方式也不一樣，在ffmpeg中通過libavformat模塊實現(xiàn)了對各個協(xié)議(file鹦筹、http、https址貌、rtsp铐拐、rtmp徘键、hls)的支持，我們這里只需調(diào)用接口av_read_frame()即可遍蟋，具體的協(xié)議實現(xiàn)細(xì)節(jié)可以暫時不用關(guān)心吹害，ffmpeg已為我們封裝好了，我們只需要在編譯的時候?qū)⑦@些協(xié)議加進(jìn)去即可虚青。

• 緩沖區(qū)

拉流模塊獲取的壓縮音視頻數(shù)據(jù)需要存放在一個緩沖區(qū)它呀，這個緩沖區(qū)稱之為壓縮數(shù)據(jù)緩沖區(qū)

拋出問題

• 1、緩沖區(qū)如何實現(xiàn)棒厘？

這里的緩沖區(qū)用于存放從拉流模塊獲取的壓縮音視頻數(shù)據(jù)纵穿，同時它還會提供給解碼模塊使用，所以我覺得這個緩沖區(qū)要滿足如下幾個要求：
1奢人、讀寫線程安全谓媒，寫線程這里就是指拉流模塊所在線程，讀線程就是指解碼模塊所在線程
2何乎、隊列空或者滿時要有讓線程休眠的機制句惯，隊列空代表沒有待解碼的數(shù)據(jù)了，這時解碼線程應(yīng)該暫停解碼支救，也就是去休眠釋放cpu抢野，隊列滿時代表壓縮數(shù)據(jù)緩沖區(qū)數(shù)據(jù)太多，這時拉流模塊應(yīng)該休眠一段時間暫緩繼續(xù)讀取數(shù)據(jù)
3搂妻、高效讀寫蒙保，隊列大小可擴展。要保證讀寫安全欲主，那么首先想到的就是鎖機制邓厕，而鎖又會帶來性能損耗，如何保證高效的讀寫呢扁瓢？

2详恼、拉流模塊如何實現(xiàn)？
首先利用ffmpeg現(xiàn)成的接口av_read_frame()函數(shù)獲取壓縮音視頻以及字幕數(shù)據(jù)包引几，該接口在libavformat模塊實現(xiàn)昧互，支持本地/遠(yuǎn)程MP4文件的讀取，以及RTMP以及RTSP協(xié)議的直播流協(xié)議伟桅，其次也要考慮如下的情況：
1敞掘、由于本地文件讀取數(shù)據(jù)非常快可能出現(xiàn)緩沖區(qū)滿的情況楣铁、遠(yuǎn)程由于網(wǎng)絡(luò)原因讀取很慢會出現(xiàn)緩沖區(qū)空的情況玖雁，那么拉流模塊要處理這兩種極端的情況
2、對回放類型還要支持暫停盖腕，重新播放赫冬，以及拖動播放
3浓镜、多線程下所有跟拉流相關(guān)的線程可能導(dǎo)致數(shù)據(jù)安全問題(其實也就是值野指針問題)

ffplay.c中緩沖區(qū)的實現(xiàn)

ffplay.c中壓縮數(shù)據(jù)緩沖區(qū)是一個用單鏈表實現(xiàn)的隊列

typedef struct MyAVPacketList {
    AVPacket pkt;
    struct MyAVPacketList *next;
    int serial; // 標(biāo)記位，1代表拉流模塊已經(jīng)準(zhǔn)備妥當(dāng)劲厌，該包可用于解碼了
} MyAVPacketList;

/** 壓縮音視頻包隊列膛薛，用鏈表實現(xiàn)
 *  疑問：為什么壓縮音視頻包隊列用隊列實現(xiàn)，而未壓縮音視頻包隊列FrameQueue卻是用數(shù)組實現(xiàn)的补鼻？
 *  分析：鏈表實現(xiàn)的隊列和數(shù)組實現(xiàn)的隊列區(qū)別就是鏈表方便擴展大小哄啄，壓縮數(shù)據(jù)包比未壓縮的要小很多，所以壓縮隊列不限制大小更適合辽幌。其次考慮到視頻幀過多的情況需要丟幀時增淹，應(yīng)該丟棄未壓縮視頻幀
 *  因為丟棄壓縮視頻中可能導(dǎo)致大量解碼出錯(比如剛好丟棄的是Idr幀)
 */
typedef struct PacketQueue {
    MyAVPacketList *first_pkt, *last_pkt;   //首尾指針
    int nb_packets; //當(dāng)前隊列的數(shù)據(jù)包個數(shù)
    int size;   // 這里是當(dāng)前隊列占用的內(nèi)存大小，并非壓縮視頻幀的大小
    int64_t duration;   // 當(dāng)前隊列所有視頻幀的總時長
    int abort_request;  // 工作結(jié)束的標(biāo)記乌企，為1時代表播放結(jié)束虑润，即將要銷毀等等
    int serial;         // 標(biāo)記位 1代表隊列是否已經(jīng)初始化并且插入了數(shù)據(jù)包，為1時隊列中的數(shù)據(jù)包才可以用于解碼
    SDL_mutex *mutex;   // 用于拉流線程和解碼線程的鎖和條件變量
    SDL_cond *cond;
} PacketQueue;

讀到這里的時候我也有個疑問加酵，為什么壓縮音視頻包隊列用單鏈表實現(xiàn)拳喻，而未壓縮音視頻包隊列FrameQueue卻是用數(shù)組實現(xiàn)的？分析：鏈表實現(xiàn)的隊列和數(shù)組實現(xiàn)的隊列區(qū)別就是鏈表方便擴展大小猪腕，由于壓縮數(shù)據(jù)包比未壓縮的要小很多冗澈，而且每個包的大小不固定，所以應(yīng)該在壓縮隊列中存儲更多的數(shù)據(jù)陋葡，其次考慮到視頻幀過多的情況需要丟幀時亚亲，應(yīng)該丟棄未壓縮視頻幀因為丟棄壓縮視頻中可能導(dǎo)致大量解碼出錯(比如剛好丟棄的是Idr幀)，綜合以上兩種因素所以用鏈表來實現(xiàn)

這里主要看兩個函數(shù)腐缤，向隊列插入數(shù)據(jù)和從隊列讀取數(shù)據(jù)

向?qū)α胁迦霐?shù)據(jù)

static int packet_queue_put_private(PacketQueue *q, AVPacket *pkt)
{
    MyAVPacketList *pkt1;

    if (q->abort_request)
       return -1;

    pkt1 = av_malloc(sizeof(MyAVPacketList));
    if (!pkt1)
        return -1;
    pkt1->pkt = *pkt;
    pkt1->next = NULL;
    if (pkt == &flush_pkt)
        q->serial++;
    pkt1->serial = q->serial;

    if (!q->last_pkt)
        q->first_pkt = pkt1;
    else
        q->last_pkt->next = pkt1;
    q->last_pkt = pkt1;
    q->nb_packets++;
    q->size += pkt1->pkt.size + sizeof(*pkt1);
    q->duration += pkt1->pkt.duration;
    /* XXX: should duplicate packet data in DV case */
    SDL_CondSignal(q->cond);
    return 0;
}
static int packet_queue_put(PacketQueue *q, AVPacket *pkt)
{
    int ret;

    SDL_LockMutex(q->mutex);
    ret = packet_queue_put_private(q, pkt);
    SDL_UnlockMutex(q->mutex);

    if (pkt != &flush_pkt && ret < 0)
        av_packet_unref(pkt);

    return ret;
}

從隊列讀取數(shù)據(jù)

/* return < 0 if aborted, 0 if no packet and > 0 if packet.  */
static int packet_queue_get(PacketQueue *q, AVPacket *pkt, int block, int *serial)
{
    MyAVPacketList *pkt1;
    int ret;

    SDL_LockMutex(q->mutex);

    for (;;) {
        if (q->abort_request) {
            ret = -1;
            break;
        }

        pkt1 = q->first_pkt;
        if (pkt1) {
            q->first_pkt = pkt1->next;
            if (!q->first_pkt)
                q->last_pkt = NULL;
            q->nb_packets--;
            q->size -= pkt1->pkt.size + sizeof(*pkt1);
            q->duration -= pkt1->pkt.duration;
            *pkt = pkt1->pkt;
            if (serial)
                *serial = pkt1->serial;
            av_free(pkt1);
            ret = 1;
            break;
        } else if (!block) {
            ret = 0;
            break;
        } else {
            SDL_CondWait(q->cond, q->mutex);
        }
    }
    SDL_UnlockMutex(q->mutex);
    return ret;
}

以上鏈表的操作還是比較容易理解的捌归，插入數(shù)據(jù)和讀取數(shù)據(jù)都會枷鎖保證線程安全

疑問：這里的緩沖單鏈表隊列采用鎖機制實現(xiàn)了線程安全，而且是全粒度的上鎖岭粤，每次上鎖均會帶來性能損耗惜索，如果實現(xiàn)一個無鎖鏈表那么效率會更高，會高多少了剃浇？

改良版無鎖隊列實現(xiàn)

待實現(xiàn)

ffplay.c中拉流模塊的實現(xiàn)

1巾兆、在main()函數(shù)啟動時調(diào)用stream_open()函數(shù)

is = stream_open(input_filename, file_iformat);
    if (!is) {
        av_log(NULL, AV_LOG_FATAL, "Failed to initialize VideoState!\n");
        do_exit(NULL);
    }

stream_open()函數(shù)的作用是創(chuàng)建VideoState結(jié)構(gòu)體對象，初始化壓縮音視頻字幕隊列PacketQueue虎囚，未壓縮音視頻音視頻字幕隊列FrameQueue角塑，然后打開拉流線程，打開拉流線程的代碼如下：

static VideoState *stream_open(const char *filename, AVInputFormat *iformat){
    ......省略代碼.....
    // 創(chuàng)建讀取線程用于讀取音視頻和字幕壓縮數(shù)據(jù)
    is->read_tid     = SDL_CreateThread(read_thread, "read_thread", is);
    if (!is->read_tid) {
        av_log(NULL, AV_LOG_FATAL, "SDL_CreateThread(): %s\n", SDL_GetError());
    .....省略代碼......
}

接下來是拉流模塊的工作代碼

static int read_thread(void *arg)
{
    VideoState *is = arg;
    AVFormatContext *ic = NULL;
    int err, i, ret;
    int st_index[AVMEDIA_TYPE_NB];
    AVPacket pkt1, *pkt = &pkt1;
    int64_t stream_start_time;
    int pkt_in_play_range = 0;
    AVDictionaryEntry *t;
    SDL_mutex *wait_mutex = SDL_CreateMutex();
    int scan_all_pmts_set = 0;
    int64_t pkt_ts;

    ......... 省略拉流相關(guān)初始化代碼.......

        /** 學(xué)習(xí)：拉流線程中當(dāng)所有的音視頻字幕隊列占用內(nèi)存超過指定MAX_QUEUE_SIZE(15M)后通過條件變量和互斥鎖等待10ms
         *  對于實時流(rtsp等等)不限制最大占用的內(nèi)存淘讥，因為畢竟是網(wǎng)絡(luò)圃伶，下載速度肯定跟不上解碼以及渲染速度，所以不可能出現(xiàn)內(nèi)存暴漲的情況适揉，但對于本地文件來說就有可能出現(xiàn)了留攒，
         *  所以這里的邏輯時針對本地文件處理的
         * infinite_buffer = 1 時代表實時流，實時流不做這樣的限制
         */
        /* if the queue are full, no need to read more */
        if (infinite_buffer<1 &&
              (is->audioq.size + is->videoq.size + is->subtitleq.size > MAX_QUEUE_SIZE
            || (stream_has_enough_packets(is->audio_st, is->audio_stream, &is->audioq) &&
                stream_has_enough_packets(is->video_st, is->video_stream, &is->videoq) &&
                stream_has_enough_packets(is->subtitle_st, is->subtitle_stream, &is->subtitleq)))) {
            /* wait 10 ms */
            SDL_LockMutex(wait_mutex);
            SDL_CondWaitTimeout(is->continue_read_thread, wait_mutex, 10);  // 當(dāng)超標(biāo)后等待10ms嫉嘀，如果10ms內(nèi)隊列中數(shù)據(jù)處理完了這里又會被喚醒炼邀，然后繼續(xù)拉流
            SDL_UnlockMutex(wait_mutex);
            continue;
        }
        if (!is->paused &&
            (!is->audio_st || (is->auddec.finished == is->audioq.serial && frame_queue_nb_remaining(&is->sampq) == 0)) &&
            (!is->video_st || (is->viddec.finished == is->videoq.serial && frame_queue_nb_remaining(&is->pictq) == 0))) {
            if (loop != 1 && (!loop || --loop)) {
                stream_seek(is, start_time != AV_NOPTS_VALUE ? start_time : 0, 0, 0);
            } else if (autoexit) {
                ret = AVERROR_EOF;
                goto fail;
            }
        }
        ret = av_read_frame(ic, pkt);
        if (ret < 0) {
            /** 學(xué)習(xí)：當(dāng)讀取到文件末尾時的處理邏輯
             *  分析：當(dāng)檢測到達(dá)文件末尾后，這里依然是通過條件變量加互斥鎖的方式讓線程休眠10ms剪侮，為什么這么做拭宁？因為播放結(jié)束后整個程序還在，這個拉流線程也沒有銷毀瓣俯，讓其休眠就不至于線程在那空轉(zhuǎn)
             */
            if ((ret == AVERROR_EOF || avio_feof(ic->pb)) && !is->eof) {
                if (is->video_stream >= 0)
                    packet_queue_put_nullpacket(&is->videoq, is->video_stream);
                if (is->audio_stream >= 0)
                    packet_queue_put_nullpacket(&is->audioq, is->audio_stream);
                if (is->subtitle_stream >= 0)
                    packet_queue_put_nullpacket(&is->subtitleq, is->subtitle_stream);
                is->eof = 1;
            }
            if (ic->pb && ic->pb->error)
                break;
            SDL_LockMutex(wait_mutex);
            SDL_CondWaitTimeout(is->continue_read_thread, wait_mutex, 10);
            SDL_UnlockMutex(wait_mutex);
            continue;
        } else {
            is->eof = 0;
        }
        
        /** 學(xué)習(xí)：拖動和(首次播放指定其實播放時間時)杰标，讀取的壓縮音視頻數(shù)據(jù)邏輯處理
         *  分析：當(dāng)首次播放指定了播放開始時間或者用戶拖動時，有可能讀取的數(shù)據(jù)包是這個起始時間之前的數(shù)據(jù)彩匕，這時候要丟棄
         *  每一個音視頻包的pts和dts都有一個起始時間參考值腔剂，這個值就是保存在AVStream的start_time變量中(這是一個固定的值，音視頻流的值一般都不一樣驼仪，這樣是為了保證音視頻的對齊)掸犬，
         *  舉例，一個視頻包的pts為10绪爸，對應(yīng)的start_time為1湾碎，那么其相對于播放時間軸的pts就為10-1 = 9，所以如果播放起始時間是2s或者用戶拖動到了2s處奠货，那么就要將9換算成時間軸在與2s比較大小
         *  如果在2s之前則將此包丟棄介褥；pkt_in_play_range就代表了這個計算過程
         */
        /* check if packet is in play range specified by user, then queue, otherwise discard */
        stream_start_time = ic->streams[pkt->stream_index]->start_time;
        pkt_ts = pkt->pts == AV_NOPTS_VALUE ? pkt->dts : pkt->pts;
        pkt_in_play_range = duration == AV_NOPTS_VALUE ||
                (pkt_ts - (stream_start_time != AV_NOPTS_VALUE ? stream_start_time : 0)) *
                av_q2d(ic->streams[pkt->stream_index]->time_base) -
                (double)(start_time != AV_NOPTS_VALUE ? start_time : 0) / 1000000
                <= ((double)duration / 1000000);
        if (pkt->stream_index == is->audio_stream && pkt_in_play_range) {
            packet_queue_put(&is->audioq, pkt);
        } else if (pkt->stream_index == is->video_stream && pkt_in_play_range
                   && !(is->video_st->disposition & AV_DISPOSITION_ATTACHED_PIC)) {
            packet_queue_put(&is->videoq, pkt);
        } else if (pkt->stream_index == is->subtitle_stream && pkt_in_play_range) {
            packet_queue_put(&is->subtitleq, pkt);
        } else {
            av_packet_unref(pkt);
        }
    }

    ret = 0;
 fail:
    if (ic && !is->ic)
        avformat_close_input(&ic);

    if (ret != 0) {
        SDL_Event event;

        event.type = FF_QUIT_EVENT;
        event.user.data1 = is;
        SDL_PushEvent(&event);
    }
    SDL_DestroyMutex(wait_mutex);
    return 0;
}

read_thread就是拉流模塊的所有代碼了，它其實是基于libavformat實現(xiàn)的拉流功能递惋，這塊代碼在ffmpeg官網(wǎng)的示例中也可以找到柔滔，這里就不多解釋了。這里只看它的一些核心代碼點丹墨。1廊遍、它是如何控制當(dāng)壓縮緩沖區(qū)超過一定大小后的處理的？2贩挣、對于本地文件讀取到文件末尾時的處理邏輯 3喉前、拖動時的處理邏輯

• 1、是如何控制當(dāng)壓縮緩沖區(qū)超過一定大小后的處理的王财？
  學(xué)習(xí)：拉流線程中當(dāng)所有的音視頻字幕隊列占用內(nèi)存超過指定MAX_QUEUE_SIZE(15M)后通過條件變量和互斥鎖等待10ms卵迂，對于實時流(rtsp等等)不限制最大占用的內(nèi)存，因為畢竟是網(wǎng)絡(luò)绒净，下載速度肯定跟不上解碼以及渲染速度见咒，所以不可能出現(xiàn)內(nèi)存暴漲的情況，但對于本地文件來說就有可能出現(xiàn)了挂疆，所以這里的邏輯時針對本地文件處理的nfinite_buffer = 1 時代表實時流改览，實時流不做這樣的限制

• 2下翎、讀取到文件末尾時的處理邏輯
  當(dāng)檢測到達(dá)文件末尾后，這里依然是通過條件變量加互斥鎖的方式讓線程休眠10ms宝当，為什么這么做视事？因為播放結(jié)束后整個程序還在，這個拉流線程也沒有銷毀庆揩，讓其休眠就不至于線程在那空轉(zhuǎn)

• 3俐东、拖動和(首次播放指定其實播放時間時)，讀取的壓縮音視頻數(shù)據(jù)邏輯處理
  當(dāng)首次播放指定了播放開始時間或者用戶拖動時订晌，有可能讀取的數(shù)據(jù)包是這個起始時間之前的數(shù)據(jù)虏辫，這時候要丟棄
  每一個音視頻包的pts和dts都有一個起始時間參考值，這個值就是保存在AVStream的start_time變量中(這是一個固定的值锈拨，音視頻流的值一般都不一樣砌庄，這樣是為了保證音視頻的對齊)，舉例奕枢，一個視頻包的pts為10鹤耍，對應(yīng)的start_time為1，那么其相對于播放時間軸的pts就為10-1 = 9验辞，所以如果播放起始時間是2s或者用戶拖動到了2s處稿黄，那么就要將9換算成時間軸在與2s比較大小，如果在2s之前則將此包丟棄跌造；pkt_in_play_range就代表了這個計算過程