前言

这里所谓的拉流从就是指从本地文件或者远程文件不停获取压缩的音视频数据包并缓存在本地待解码的过程，用一张图形象的画出来其过程如下：

image.png

• 拉流模块

这里要有个拉流线程让拉流模块在此线程中不停的工作，它需要满足忙时工作闲时休眠等待，对于拉流模块，在ffmpeg的世界里也可以通俗的称为解析器，不同的协议从其中获取数据的方式也不一样，在ffmpeg中通过libavformat模块实现了对各个协议(file、http、https、rtsp、rtmp、hls)的支持，我们这里只需调用接口av_read_frame()即可，具体的协议实现细节可以暂时不用关心，ffmpeg已为我们封装好了，我们只需要在编译的时候将这些协议加进去即可。

• 缓冲区

拉流模块获取的压缩音视频数据需要存放在一个缓冲区，这个缓冲区称之为压缩数据缓冲区

抛出问题

• 1、缓冲区如何实现？

这里的缓冲区用于存放从拉流模块获取的压缩音视频数据，同时它还会提供给解码模块使用，所以我觉得这个缓冲区要满足如下几个要求：
1、读写线程安全，写线程这里就是指拉流模块所在线程，读线程就是指解码模块所在线程
2、队列空或者满时要有让线程休眠的机制，队列空代表没有待解码的数据了，这时解码线程应该暂停解码，也就是去休眠释放cpu，队列满时代表压缩数据缓冲区数据太多，这时拉流模块应该休眠一段时间暂缓继续读取数据
3、高效读写，队列大小可扩展。要保证读写安全，那么首先想到的就是锁机制，而锁又会带来性能损耗，如何保证高效的读写呢？

2、拉流模块如何实现？
首先利用ffmpeg现成的接口av_read_frame()函数获取压缩音视频以及字幕数据包，该接口在libavformat模块实现，支持本地/远程MP4文件的读取，以及RTMP以及RTSP协议的直播流协议，其次也要考虑如下的情况：
1、由于本地文件读取数据非常快可能出现缓冲区满的情况、远程由于网络原因读取很慢会出现缓冲区空的情况，那么拉流模块要处理这两种极端的情况
2、对回放类型还要支持暂停，重新播放，以及拖动播放
3、多线程下所有跟拉流相关的线程可能导致数据安全问题(其实也就是值野指针问题)

ffplay.c中缓冲区的实现

ffplay.c中压缩数据缓冲区是一个用单链表实现的队列

typedef struct MyAVPacketList {
    AVPacket pkt;
    struct MyAVPacketList *next;
    int serial; // 标记位，1代表拉流模块已经准备妥当，该包可用于解码了
} MyAVPacketList;

/** 压缩音视频包队列，用链表实现
 *  疑问：为什么压缩音视频包队列用队列实现，而未压缩音视频包队列FrameQueue却是用数组实现的？
 *  分析：链表实现的队列和数组实现的队列区别就是链表方便扩展大小，压缩数据包比未压缩的要小很多，所以压缩队列不限制大小更适合。其次考虑到视频帧过多的情况需要丢帧时，应该丢弃未压缩视频帧
 *  因为丢弃压缩视频中可能导致大量解码出错(比如刚好丢弃的是Idr帧)
 */
typedef struct PacketQueue {
    MyAVPacketList *first_pkt, *last_pkt;   //首尾指针
    int nb_packets; //当前队列的数据包个数
    int size;   // 这里是当前队列占用的内存大小，并非压缩视频帧的大小
    int64_t duration;   // 当前队列所有视频帧的总时长
    int abort_request;  // 工作结束的标记，为1时代表播放结束，即将要销毁等等
    int serial;         // 标记位 1代表队列是否已经初始化并且插入了数据包，为1时队列中的数据包才可以用于解码
    SDL_mutex *mutex;   // 用于拉流线程和解码线程的锁和条件变量
    SDL_cond *cond;
} PacketQueue;

读到这里的时候我也有个疑问，为什么压缩音视频包队列用单链表实现，而未压缩音视频包队列FrameQueue却是用数组实现的？分析：链表实现的队列和数组实现的队列区别就是链表方便扩展大小，由于压缩数据包比未压缩的要小很多，而且每个包的大小不固定，所以应该在压缩队列中存储更多的数据，其次考虑到视频帧过多的情况需要丢帧时，应该丢弃未压缩视频帧因为丢弃压缩视频中可能导致大量解码出错(比如刚好丢弃的是Idr帧)，综合以上两种因素所以用链表来实现

这里主要看两个函数，向队列插入数据和从队列读取数据

向对列插入数据

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

从队列读取数据

/* 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;
}

以上链表的操作还是比较容易理解的，插入数据和读取数据都会枷锁保证线程安全

疑问：这里的缓冲单链表队列采用锁机制实现了线程安全，而且是全粒度的上锁，每次上锁均会带来性能损耗，如果实现一个无锁链表那么效率会更高，会高多少了？

改良版无锁队列实现

待实现

ffplay.c中拉流模块的实现

1、在main()函数启动时调用stream_open()函数

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()函数的作用是创建VideoState结构体对象，初始化压缩音视频字幕队列PacketQueue，未压缩音视频音视频字幕队列FrameQueue，然后打开拉流线程，打开拉流线程的代码如下：

static VideoState *stream_open(const char *filename, AVInputFormat *iformat){
    ......省略代码.....
    // 创建读取线程用于读取音视频和字幕压缩数据
    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;

    ......... 省略拉流相关初始化代码.......

        /** 学习：拉流线程中当所有的音视频字幕队列占用内存超过指定MAX_QUEUE_SIZE(15M)后通过条件变量和互斥锁等待10ms
         *  对于实时流(rtsp等等)不限制最大占用的内存，因为毕竟是网络，下载速度肯定跟不上解码以及渲染速度，所以不可能出现内存暴涨的情况，但对于本地文件来说就有可能出现了，
         *  所以这里的逻辑时针对本地文件处理的
         * 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);  // 当超标后等待10ms，如果10ms内队列中数据处理完了这里又会被唤醒，然后继续拉流
            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) {
            /** 学习：当读取到文件末尾时的处理逻辑
             *  分析：当检测到达文件末尾后，这里依然是通过条件变量加互斥锁的方式让线程休眠10ms，为什么这么做？因为播放结束后整个程序还在，这个拉流线程也没有销毁，让其休眠就不至于线程在那空转
             */
            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;
        }
        
        /** 学习：拖动和(首次播放指定其实播放时间时)，读取的压缩音视频数据逻辑处理
         *  分析：当首次播放指定了播放开始时间或者用户拖动时，有可能读取的数据包是这个起始时间之前的数据，这时候要丢弃
         *  每一个音视频包的pts和dts都有一个起始时间参考值，这个值就是保存在AVStream的start_time变量中(这是一个固定的值，音视频流的值一般都不一样，这样是为了保证音视频的对齐)，
         *  举例，一个视频包的pts为10，对应的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实现的拉流功能，这块代码在ffmpeg官网的示例中也可以找到，这里就不多解释了。这里只看它的一些核心代码点。1、它是如何控制当压缩缓冲区超过一定大小后的处理的？2、对于本地文件读取到文件末尾时的处理逻辑 3、拖动时的处理逻辑

• 1、是如何控制当压缩缓冲区超过一定大小后的处理的？
  学习：拉流线程中当所有的音视频字幕队列占用内存超过指定MAX_QUEUE_SIZE(15M)后通过条件变量和互斥锁等待10ms，对于实时流(rtsp等等)不限制最大占用的内存，因为毕竟是网络，下载速度肯定跟不上解码以及渲染速度，所以不可能出现内存暴涨的情况，但对于本地文件来说就有可能出现了，所以这里的逻辑时针对本地文件处理的nfinite_buffer = 1 时代表实时流，实时流不做这样的限制

• 2、读取到文件末尾时的处理逻辑
  当检测到达文件末尾后，这里依然是通过条件变量加互斥锁的方式让线程休眠10ms，为什么这么做？因为播放结束后整个程序还在，这个拉流线程也没有销毁，让其休眠就不至于线程在那空转

• 3、拖动和(首次播放指定其实播放时间时)，读取的压缩音视频数据逻辑处理
  当首次播放指定了播放开始时间或者用户拖动时，有可能读取的数据包是这个起始时间之前的数据，这时候要丢弃
  每一个音视频包的pts和dts都有一个起始时间参考值，这个值就是保存在AVStream的start_time变量中(这是一个固定的值，音视频流的值一般都不一样，这样是为了保证音视频的对齐)，举例，一个视频包的pts为10，对应的start_time为1，那么其相对于播放时间轴的pts就为10-1 = 9，所以如果播放起始时间是2s或者用户拖动到了2s处，那么就要将9换算成时间轴在与2s比较大小，如果在2s之前则将此包丢弃；pkt_in_play_range就代表了这个计算过程