FFmepg-- 32-ffplay源码- PacketQueue 的线程安全机制 以及 serial 字段的作用

一 PacketQueue 的线程安全设计

在 ffplay.c 中，PacketQueue 是一个典型的生产者-消费者队列：

生产者：read_thread（从文件/网络读取 AVPacket 并放入队列）
消费者：解码线程（如 audio_thread / video_thread）从队列中取出 AVPacket 解码

线程同步手段

typedefstructPacketQueue{AVPacketList*first_pkt,*last_pkt;intnb_packets;// 当前包数量intsize;// 总字节数（用于限流）int64_tduration;// 总时长（毫秒）intabort_request;// 是否请求终止intserial;// 👈 关键字段：播放序列号SDL_mutex*mutex;// 互斥锁SDL_cond*cond;// 条件变量}PacketQueue;

SDL_mutex：保护对队列结构体（如 first_pkt, last_pkt, nb_packets 等）的并发访问。
SDL_cond：用于阻塞/唤醒：
消费者调用 packet_queue_get(…, block=1) 时，若队列为空，则 SDL_CondWait(cond, mutex) 阻塞；
生产者调用 packet_queue_put() 后，调用 SDL_CondSignal(cond) 唤醒等待的消费者。

二 serial 字段的作用详解

为什么需要 serial？

当用户执行 seek（快进/快退）操作时，旧的 AVPacket 已经无效，必须被丢弃。此时可能出现以下情况：

read_thread 可能还在往队列里塞旧数据；
解码线程可能还在处理旧数据；
如果不清除这些“过期”数据，就会出现：
音频“回放杂音”
视频跳帧混乱
音画不同步

serial 用于标识“当前播放上下文”的版本号。

serial 的工作机制

初始化时：q->serial = 0

执行 seek 时：
调用packet_queue_flush(&is->audioq)清空队列；
插入一个特殊的flush_pkt（其data == NULL）；
执行q->serial++（例如从 0 → 1）

此后所有新入队的 packet 都会设置pkt->serial = q->serial

解码线程在取到 packet 后，会检查：

if(pkt->serial!=decoder->pkt_serial){av_packet_unref(pkt);continue;// 丢弃旧序列数据}

其中decoder->pkt_serial会在 flush 后被更新为新的 serial。

三 简化版示例代码

以下是一个高度简化但功能完整的PacketQueue实现，突出serial和线程安全逻辑：

#include<SDL2/SDL.h>#include<libavcodec/avcodec.h>#defineMAX_QUEUE_SIZE(15*1024*1024)typedefstructMyAVPacketList{AVPacket pkt;intserial;structMyAVPacketList*next;}MyAVPacketList;typedefstructPacketQueue{MyAVPacketList*first,*last;intnb_packets;intsize;int64_tduration;intabort_request;intserial;// 序列号SDL_mutex*mutex;SDL_cond*cond;}PacketQueue;voidpacket_queue_init(PacketQueue*q){memset(q,0,sizeof(PacketQueue));q->mutex=SDL_CreateMutex();q->cond=SDL_CreateCond();q->serial=0;}intpacket_queue_put(PacketQueue*q,AVPacket*pkt){MyAVPacketList*pkt1;SDL_LockMutex(q->mutex);if(q->abort_request){SDL_UnlockMutex(q->mutex);return-1;}pkt1=av_malloc(sizeof(MyAVPacketList));if(!pkt1)gotofail;pkt1->pkt=*pkt;pkt1->serial=q->serial;// 绑定当前 serialpkt1->next=NULL;if(!q->last)q->first=pkt1;elseq->last->next=pkt1;q->last=pkt1;q->nb_packets++;q->size+=pkt1->pkt.size+sizeof(*pkt1);q->duration+=pkt1->pkt.duration;SDL_CondSignal(q->cond);// 唤醒消费者SDL_UnlockMutex(q->mutex);return0;fail:SDL_UnlockMutex(q->mutex);return-1;}// 获取 packet，block=1 表示阻塞等待intpacket_queue_get(PacketQueue*q,AVPacket*pkt,intblock,int*serial){MyAVPacketList*pkt1;intret;SDL_LockMutex(q->mutex);for(;;){if(q->abort_request){ret=-1;break;}pkt1=q->first;if(pkt1){q->first=pkt1->next;if(!q->first)q->last=NULL;q->nb_packets--;q->size-=pkt1->pkt.size+sizeof(*pkt1);q->duration-=pkt1->pkt.duration;*pkt=pkt1->pkt;if(serial)*serial=pkt1->serial;// 返回 packet 的 serialav_free(pkt1);ret=1;break;}elseif(!block){ret=0;break;}else{SDL_CondWait(q->cond,q->mutex);// 阻塞等待}}SDL_UnlockMutex(q->mutex);returnret;}// seek 时调用：清空队列 + serial++voidpacket_queue_flush(PacketQueue*q){MyAVPacketList*pkt,*pkt1;SDL_LockMutex(q->mutex);for(pkt=q->first;pkt;pkt=pkt1){pkt1=pkt->next;av_packet_unref(&pkt->pkt);av_free(pkt);}q->first=q->last=NULL;q->nb_packets=0;q->size=0;q->duration=0;q->serial++;// 关键：序列号递增SDL_UnlockMutex(q->mutex);}

使用场景（解码线程伪代码）

intwanted_serial=is->audioq.serial;// 期望的 serialwhile(1){intserial;AVPacket pkt;if(packet_queue_get(&is->audioq,&pkt,1,&serial)<0)break;if(serial!=wanted_serial){av_packet_unref(&pkt);continue;// 丢弃旧序列数据}// 正常解码...decode_audio(&pkt);av_packet_unref(&pkt);}

Seek 发生时

// 用户 seek 到新位置packet_queue_flush(&is->audioq);// serial 自增packet_queue_flush(&is->videoq);// read_thread 会重新开始读取，并给新 packet 打上新 serial

四 总结