前言前面写了 RTMP 推流和拉流客户端这篇补齐服务端。服务端是整条链路的枢纽——它同时收 RTMP 推流、RTSP 推流然后通过 Muxer 把同一路流分发给不同协议的拉流客户端。本文围绕RtmpSession服务端单个 RTMP 会话展开每个协议步骤都给出真实十六进制字节 中文逐字段注释。读完你能掌握RTMP 服务端三次握手的字节交换connect / createStream / play / publish 的处理推流方向如何解析 avcC/ASC 并构建 Track拉流方向如何生成 AVCDecoderConfigurationRecordRTMP 推 → RTSP 拉的协议桥接机制一、RtmpSession 定位RtmpSession是一个 TCP 连接对应一个实例同时支持两个方向方向触发数据流拉流客户端发playMediaSource → RtmpMuxer → 客户端推流客户端发publish客户端 → 解析 → MediaSource一个类共用两个方向通过_playing/_publishing两个标志区分。整体数据流RTMP 客户端推流 ──► RtmpSession::handlePublish │ ├─ 解析 avcC → VideoTrackextradata AnnexB SPSPPS ├─ 解析 ASC → AudioTrackextradata ASC └─ MediaSource::writeFrame(frame) │ ▼ RingBuffer │ 分发 ┌───────────┴────────────┐ ▼ ▼ RtspMuxer RtmpMuxer │ │ ▼ ▼ RTSP 客户端拉流 RTMP 客户端拉流二、三次握手2.1 客户端 → 服务器C0 C1字节序列┌──────────────────────────────────────────────────────────────────┐ │ 03 ← C0: 版本号 0x03 │ ├──────────────────────────────────────────────────────────────────┤ │ 00 00 00 00 ← C1 前 4 字节客户端时间戳ms │ │ 00 00 00 00 ← C1 第 5~8 字节保留零 │ │ 8D 4C B6 2E 91 0F 5A C3 ... ← C1 第 9~1536 字节随机 1528 字节 │ │ ...共 1536 字节 │ └──────────────────────────────────────────────────────────────────┘中文说明字段大小含义C0.version1B固定 0x03C1.timestamp4B客户端启动时间戳毫秒C1.zero4B固定 0x00000000C1.random1528B随机字节2.2 服务器 → 客户端S0 S1 S2QByteArray makeServerS0S1S2(const QByteArray c1) { QByteArray out; out.append(char(0x03)); // S0 QByteArray s1(1536, \0); for (int i 8; i 1536; i) s1[i] char(i 0xFF); out.append(s1); // S1服务器数据 out.append(c1); // S2 C1 回显简单握手 return out; }字节序列03 ← S0: 版本号 0x03 00 00 00 00 ← S1 前 4 字节服务器时间戳 00 00 00 00 ← S1 第 5~8 字节保留零 08 09 0A 0B 0C 0D 0E 0F ... ← S1 第 9~1536 字节随机数据 ...共 1536 字节 C1 的完整 1536 字节回显 ← S2 客户端的 C1总长度1 1536 1536 3073 字节2.3 客户端 → 服务器C2C1 之后的 S1 拷贝1536 字节服务器不校验 C2握手完成。2.4 服务端处理代码void RtmpSession::processHandshake() { constexpr int c0c1Size 1 1536; // C0(1) C1(1536) if (_hsBuf.size() c0c1Size) return; // 数据不足等下次 if (uint8_t(_hsBuf[0]) ! 0x03) { // 版本号必须 0x03 _sock-abort(); // 非法直接断开 return; } QByteArray c1 _hsBuf.mid(1, 1536); // 提取 C1 _sock-write(makeServerS0S1S2(c1)); // 回 S0S1S2 _hsBuf.remove(0, c0c1Size); // 消费 C0C1 _hsDone true; // 标记完成 if (_hsBuf.size() 1536) // 若 C2 也到了 _hsBuf.remove(0, 1536); // 一并消费 }三、connect 命令3.1 客户端 → 服务器RTMP 消息结构03 ← Basic Header: fmt0, csid3 00 00 00 ← timestamp 0 00 00 C0 ← length 192 (0xC0) 14 ← typeId 20 (MSG_CMD) 00 00 00 00 ← streamId 0 (小端) ├──────────────────────────────────────────────────────────┤ │ 02 00 07 63 6F 6E 6E 65 63 74 │ String connect │ 00 3F F0 00 00 00 00 00 00 │ Number 1.0 (txn) │ 03 │ Object 开始 │ 00 03 61 70 70 02 00 04 6C 69 76 65 │ key app → String live │ 00 05 74 63 55 72 6C │ key tcUrl │ 02 00 1D 72 74 6D 70 3A 2F 2F ... │ → rtmp://127.0.0.1:1935/live │ 00 08 66 6C 61 73 68 56 65 72 │ key flashVer │ 02 00 0C 46 4D 4C 45 2F 33 2E 30 │ → FMLE/3.0 │ 00 0B 61 75 64 69 6F 43 6F 64 65 63 73 │ key audioCodecs │ 00 40 B0 00 00 00 00 00 00 │ → Number 1024 (AAC) │ 00 0B 76 69 64 65 6F 43 6F 64 65 63 73 │ key videoCodecs │ 00 40 80 00 00 00 00 00 00 │ → Number 128 (H.264) │ 00 00 09 │ Object 结束 └──────────────────────────────────────────────────────────┘3.2 服务端解析void RtmpSession::handleConnect(double txn, AmfDecoder dec) { _connectTxn txn; // 保存事务号 AmfValue obj dec.readValue(); // 读参数对象 _app obj.value(app).asString(); // 取出 app live sendConnectResult(txn); // 回复 _result }3.3 服务器 → 客户端_result字节序列03 00 00 00 00 00 B8 14 00 00 00 00 ← Chunk Msg 头 (length184) 02 00 07 5F 72 65 73 75 6C 74 ← String _result 00 3F F0 00 00 00 00 00 00 ← Number 1.0 (txn) 05 ← Null 03 ← Object 1: 服务器信息 00 06 66 6D 73 56 65 72 ← key fmsVer 02 00 0F 46 4D 53 2F 33 2C 30 2C 31 2C 31 32 33 ← FMS/3,0,1,123 00 0C 63 61 70 61 62 69 6C 69 74 69 65 73 ← key capabilities 00 40 3F 80 00 00 00 00 00 ← Number 31 00 00 09 ← Object 1 结束 03 ← Object 2: 结果信息 00 05 6C 65 76 65 6C ← key level 02 00 06 73 74 61 74 75 73 ← status 00 04 63 6F 64 65 ← key code 02 00 1F 4E 65 74 43 6F 6E 6E 65 63 74 69 6F 6E 2E 43 6F 6E 6E 65 63 74 2E 53 75 63 63 65 73 73 ← NetConnection.Connect.Success 00 0B 64 65 73 63 72 69 70 74 69 6F 6E ← key description 02 00 15 43 6F 6E 6E 65 63 74 69 6F 6E 20 73 75 63 63 65 65 64 65 64 2E ← Connection succeeded. 00 00 09 ← Object 2 结束中文说明字段值含义_result命令名响应1.0事务号与 connect 对应Object 1fmsVer / capabilities服务器信息Object 2levelstatus /codeNetConnection.Connect.Success★ 成功标志Object 2description描述3.4 服务端代码void RtmpSession::sendConnectResult(double txn) { AmfEncoder enc; enc.writeString(_result); // 命令名 enc.writeNumber(txn); // 事务号 AmfValue cmdObj AmfValue::makeObject(); cmdObj.set(fmsVer, AmfValue::makeString(FMS/3,0,1,123)); cmdObj.set(capabilities, AmfValue::makeNumber(31)); enc.writeObject(cmdObj); AmfValue info AmfValue::makeObject(); info.set(level, AmfValue::makeString(status)); info.set(code, AmfValue::makeString(NetConnection.Connect.Success)); info.set(description, AmfValue::makeString(Connection succeeded.)); info.set(objectEncoding, AmfValue::makeNumber(0)); enc.writeObject(info); writeRtmpMessage(MSG_CMD, 0, CHUNK_SYSTEM, enc.data()); // 顺带发 Window Ack Size 5MB uint8_t buf[4]; ByteOrder::set_be32(buf, 5000000); writeRtmpMessage(MSG_WIN_SIZE, 0, CHUNK_NETWORK, QByteArray(reinterpret_castconst char*(buf), 4)); }四、createStream 命令4.1 客户端 → 服务器03 00 00 00 00 00 19 14 00 00 00 00 ← Chunk Msg 头 (length25) 02 00 0C 63 72 65 61 74 65 53 74 72 65 61 6D ← String createStream 00 40 00 00 00 00 00 00 00 ← Number 2.0 (txn) 05 ← Null4.2 服务器 → 客户端03 00 00 00 00 00 15 14 00 00 00 00 ← Chunk Msg 头 (length21) 02 00 07 5F 72 65 73 75 6C 74 ← _result 00 40 00 00 00 00 00 00 00 ← txn 2.0 05 ← Null 00 3F F0 00 00 00 00 00 00 ← ★ streamId 1.04.3 服务端代码void RtmpSession::handleCreateStream(double txn) { _createStreamTxn txn; sendCreateStreamResult(txn); } void RtmpSession::sendCreateStreamResult(double txn) { AmfEncoder enc; enc.writeString(_result); enc.writeNumber(txn); enc.writeNull(); enc.writeNumber(1); // ★ streamId 1简化版写死 _streamId 1; writeRtmpMessage(MSG_CMD, 0, CHUNK_SYSTEM, enc.data()); }⚠️ 写死 streamId 的隐患同一连接多次createStream会返回相同 ID。生产环境应改成自增static std::atomicuint32_t gNextStreamId{1}; _streamId gNextStreamId;五、拉流方向play5.1 客户端 → 服务器03 00 00 00 00 00 20 14 01 00 00 00 ← ★ streamId 1小端 02 00 04 70 6C 61 79 ← String play 00 40 08 00 00 00 00 00 00 ← txn 3.0 05 ← Null 02 00 04 74 65 73 74 ← String test流名 00 C0 9F 40 00 00 00 00 00 ← Number -2000.0负数live5.2 服务端处理流程void RtmpSession::handlePlay(double txn, const QString name) { QString streamName extractStreamName(name); // test MediaTuple tuple; tuple.schema rtmp; // 先按 rtmp 找 tuple.vhost 127.0.0.1; tuple.app _app; // live tuple.stream streamName; // test // ★ 协议桥接先按 rtmp 找找不到换 rtsp _playSource MediaSourceManager::instance().find(tuple); if (!_playSource) { MediaTuple rtspTuple tuple; rtspTuple.schema rtsp; _playSource MediaSourceManager::instance().find(rtspTuple); } if (!_playSource) { sendPlayOnStatus(NetStream.Play.StreamNotFound, Stream not found); return; } // ★ 从 StreamManager 拿共享 Muxer没有就现场创建 auto muxer std::dynamic_pointer_castRtmpMuxer( StreamManager::instance().getRtmpMuxer(_playSource)); if (!muxer) { muxer std::make_sharedRtmpMuxer(_playSource); muxer-start(); StreamManager::instance().registerRtmpMuxer(_playSource, muxer); } _rtmpMuxer muxer; _rtmpMuxer-addSession(this); // ★ 加入 Muxer 分发列表 // 回复三段 onStatus sendPlayOnStatus(NetStream.Play.Start, Started playing.); sendPlayOnStatus(NetStream.Data.Start, ); sendPlayOnStatus(NetStream.Play.Reset, ); // 发 onMetaData config sendOnMetaData(_playSource); sendVideoConfig(_playSource); sendAudioConfig(_playSource); _playing true; }5.3 服务器 → 客户端三段 onStatusonStatus (Play.Reset)03 00 00 00 00 00 7A 14 01 00 00 00 ← streamId 1 02 00 08 6F 6E 53 74 61 74 75 73 ← onStatus 00 00 00 00 00 00 00 00 00 ← txn 0 05 ← Null 03 ... 00 00 09 ← Object level status code NetStream.Play.Reset description Playing and resetting.onStatus (Play.Start)code NetStream.Play.Start ← ★ 播放开始标志 description Started playing.onStatus (Data.Start)code NetStream.Data.Start description 5.4 服务器 → 客户端onMetaData字节序列03 00 00 00 00 01 06 12 01 00 00 00 ← Chunk Msg 头 (length262) 02 00 0D 40 73 65 74 44 61 74 61 46 72 61 6D 65 ← String setDataFrame 02 00 0A 6F 6E 4D 65 74 61 44 61 74 61 ← String onMetaData 08 ← ECMA Array 00 00 00 07 ← count 7 00 08 64 75 72 61 74 69 6F 6E ← key duration 00 00 00 00 00 00 00 00 ← 0 00 05 77 69 64 74 68 ← key width 00 40 88 00 00 00 00 00 00 ← 1920 00 06 68 65 69 67 68 74 ← key height 00 40 87 00 00 00 00 00 00 ← 1080 00 09 66 72 61 6D 65 72 61 74 65 ← key framerate 00 40 39 00 00 00 00 00 00 ← 25 00 0C 76 69 64 65 6F 63 6F 64 65 63 69 64 ← key videocodecid 00 40 1C 00 00 00 00 00 00 ← 7 (H.264) 00 0C 61 75 64 69 6F 63 6F 64 65 63 69 64 ← key audiocodecid 00 40 24 00 00 00 00 00 00 ← 10 (AAC) 00 00 09 ← 结束5.5 服务器 → 客户端Video Sequence Header这一步最关键——服务端要把 MediaSource 里存的内存的 AnnexB extradata 转成AVCDecoderConfigurationRecord再发出去。字节序列05 00 00 00 00 00 2A 09 01 00 00 00 ← streamId 1 17 ← Video Tag Header (key AVC) 00 ← AVCPacketType 0 (Sequence Header) 00 00 00 ← CompositionTime 01 ← ★ configurationVersion 1 42 C0 1E ← profile / compat / level FF ← reserved(6) | lengthSizeMinusOne3 E1 ← reserved(3) | numOfSPS1 00 19 ← SPS 长度 25 67 42 C0 1E D9 00 50 05 BB 01 10 00 00 03 00 10 00 00 03 03 C8 F1 22 6A 2C ← SPS 数据 01 ← numOfPPS 1 00 05 ← PPS 长度 5 68 CE 3C 80 ← PPS 数据服务端代码void RtmpSession::sendVideoConfig(const MediaSource::Ptr src) { auto vt std::dynamic_pointer_castVideoTrack(src-findVideoTrack()); QByteArray extra vt-extradata(); // AnnexB SPSPPS // 从 AnnexB 提取 SPS/PPS QListQByteArray spsList, ppsList; const uint8_t* d reinterpret_castconst uint8_t*(extra.constData()); int size extra.size(), pos 0, start -1; while (pos size) { bool atEnd (pos size); bool sc3 (!atEnd pos3 size d[pos]0 d[pos1]0 d[pos2]1); bool sc4 (!atEnd pos4 size d[pos]0 d[pos1]0 d[pos2]0 d[pos3]1); if (atEnd || sc3 || sc4) { if (start 0 pos start) { QByteArray nalu(reinterpret_castconst char*(d start), pos - start); uint8_t t uint8_t(nalu[0]) 0x1F; if (t 7) spsList.append(nalu); else if (t 8) ppsList.append(nalu); } if (atEnd) break; pos sc4 ? 4 : 3; start pos; } else pos; } // 拼 AVCDecoderConfigurationRecord QByteArray cfg; cfg.append(char(1)); // configurationVersion cfg.append(sps[1]); cfg.append(sps[2]); cfg.append(sps[3]); cfg.append(char(0xFF)); // lengthSizeMinusOne3 cfg.append(char(0xE1)); // numOfSPS1 uint8_t lb[2]; ByteOrder::set_be16(lb, uint16_t(sps.size())); cfg.append(reinterpret_castconst char*(lb), 2); cfg.append(sps); cfg.append(char(0x01)); // numOfPPS1 ByteOrder::set_be16(lb, uint16_t(pps.size())); cfg.append(reinterpret_castconst char*(lb), 2); cfg.append(pps); // 拼 RTMP video tag QByteArray body; body.append(char(0x17)); // key AVC body.append(char(0x00)); // Sequence Header body.append(3, \0); // CompositionTime body.append(cfg); writeRtmpMessage(MSG_VIDEO, _streamId, CHUNK_VIDEO, body, 0); }5.6 服务器 → 客户端Audio Sequence Header字节序列04 00 00 00 00 00 04 08 01 00 00 00 AF ← Audio Tag Header (AAC/44k/16/stereo) 00 ← AAC sequence header 12 10 ← AudioSpecificConfigASC 解析12 100001 0010 0001 0000audioObjectType 2 (AAC-LC) samplingFreqIdx 4 (44100 Hz) channelConfig 2 (stereo)5.7 服务器 → 客户端媒体帧由RtmpMuxer从 MediaSource 拿到 Frame 后转发bool RtmpMuxer::onTrackFrame(const Frame::Ptr frame) { if (frame-isVideo()) { QByteArray body makeVideoBody(frame); // AnnexB → AVCC dispatch(MSG_VIDEO, body, videoTs(frame)); } // ... }视频帧字节05 00 03 E8 00 0A B6 09 01 00 00 00 ← timestamp 1000ms 17 ← key AVC 01 ← NALU 00 00 00 ← CompositionTime 00 00 0A B2 ← 第一个 NALU 长度 65 88 84 ... ← IDR NALU ...六、推流方向publish6.1 客户端 → 服务器03 00 00 00 00 00 20 14 01 00 00 00 ← streamId 1 02 00 07 70 75 62 6C 69 73 68 ← String publish 00 40 08 00 00 00 00 00 00 ← txn 3.0 05 ← Null 02 00 04 74 65 73 74 ← String test流名 02 00 04 6C 69 76 65 ← String live推流类型6.2 服务端处理void RtmpSession::handlePublish(double txn, const QString name, const QString type) { _publishTxn txn; _publishType type; // live _streamName extractStreamName(name); MediaTuple tuple; tuple.schema rtmp; tuple.vhost 127.0.0.1; tuple.app _app; tuple.stream _streamName; _publishSource std::make_sharedMediaSource(tuple); // 创建 MediaSource _publishRegistered false; _pubVideoTrack.reset(); _pubAudioTrack.reset(); _videoConfigReady false; _audioConfigReady false; _pubBaseDts -1; // 音视频共享 DTS 基准 sendPublishOnStatus(NetStream.Publish.Start, Started publishing.); _publishing true; }6.3 服务器响应03 00 00 00 00 00 XX 14 01 00 00 00 02 00 08 6F 6E 53 74 61 74 75 73 00 00 00 00 00 00 00 00 00 05 03 ... 00 00 09 level status code NetStream.Publish.Start description Started publishing.6.4 客户端 → 服务器onMetaData和拉流方向一样但是客户端发给服务端03 00 00 00 00 01 06 12 01 00 00 00 02 00 0D setDataFrame 02 00 0A onMetaData 08 00 00 00 07 00 08 duration 00 00 00 00 00 00 00 00 00 05 width 00 40 88 00 00 00 00 00 00 ← 1920 00 06 height 00 40 87 00 00 00 00 00 00 ← 1080 00 09 framerate 00 40 39 00 00 00 00 00 00 ← 25 00 0C videocodecid 00 40 1C 00 00 00 00 00 00 ← 7 00 0C audiocodecid 00 40 24 00 00 00 00 00 00 ← 10 00 00 09服务端缓存元数据void RtmpSession::handleData(const RtmpPacket::Ptr pkt) { AmfDecoder dec(Buffer(pkt-body)); QString first dec.readString(); // setDataFrame QString metaName first; if (first setDataFrame) { metaName dec.readString(); // onMetaData } if (metaName ! onMetaData) return; AmfValue meta dec.readValue(); _meta.width meta.value(width).asInt(); _meta.height meta.value(height).asInt(); _meta.fps meta.value(framerate).asInt(); _meta.sampleRate meta.value(audiosamplerate).asInt(); _meta.channels meta.value(stereo).asBool() ? 2 : 1; _meta.valid true; }6.5 客户端 → 服务器Video Sequence Header05 00 00 00 00 00 2A 09 01 00 00 00 17 ← key AVC 00 ← Sequence Header 00 00 00 ← CompositionTime 01 ← configurationVersion 42 C0 1E FF E1 00 19 ← 头 67 42 C0 1E D9 ... 25 字节 SPS ← SPS 01 00 05 68 CE 3C 80 ← PPS服务端处理void RtmpSession::handlePublishVideo(const RtmpPacket::Ptr pkt) { const uint8_t* p (const uint8_t*)pkt-body.constData(); uint8_t frameType (p[0] 4) 0x0F; uint8_t codecNum p[0] 0x0F; // 7 AVC uint8_t avcType p[1]; // 0 Sequence Header if (codecNum ! 7) return; // 建立 DTS 基准 if (_pubBaseDts 0) _pubBaseDts int64_t(pkt-timeStamp); uint64_t dts pkt-timeStamp - _pubBaseDts; if (avcType 0) { QByteArray avcCfg pkt-body.mid(5); // ★ avcC → AnnexB SPSPPS int naluLenSize 4; QByteArray annexBExtra avcConfigToAnnexB(avcCfg, naluLenSize); if (!_pubVideoTrack) { _pubVideoTrack std::make_sharedVideoTrack(CodecId::H264, 0); _pubVideoTrack-setWidth(_meta.width); _pubVideoTrack-setHeight(_meta.height); _pubVideoTrack-setFps(_meta.fps); _pubVideoTrack-setNaluLengthSize(naluLenSize); _pubVideoTrack-setExtradata(annexBExtra); // AnnexB 存入 extradata _publishSource-addTrack(_pubVideoTrack); } _videoConfigReady true; // config 帧写入 MediaSource关键帧标记供 RingBuffer GOP 使用 auto frame FrameFactory::create( CodecId::H264, annexBExtra, 0, 0, false, true, 0); _publishSource-writeFrame(frame, false); finalizePublishSource(); // 检查所有 track 是否 ready return; } }avcC → AnnexB 代码static QByteArray avcConfigToAnnexB(const QByteArray cfg, int naluLenSize) { // cfg 结构[version][profile][compat][level][FF][E1][spsLen][sps][01][ppsLen][pps] naluLenSize (p[4] 0x03) 1; // 从 [4] 读长度字段宽度 int off 5; int numSps p[off] 0x1F; // numOfSPS QByteArray out; // 遍历 SPS for (int i 0; i numSps; i) { int len (p[off] 8) | p[off1]; off 2; out.append(\x00\x00\x00\x01, 4); // 起始码 out.append((const char*)(p off), len); // SPS 数据 off len; } // 遍历 PPS int numPps p[off]; for (int i 0; i numPps; i) { int len (p[off] 8) | p[off1]; off 2; out.append(\x00\x00\x00\x01, 4); out.append((const char*)(p off), len); off len; } return out; }输出00 00 00 01 67 42 C0 1E ...SPS 00 00 00 01 68 CE 3C 80 PPS6.6 客户端 → 服务器Video FrameAVCC05 00 03 E8 00 0A B6 09 01 00 00 00 17 ← key AVC 01 ← NALU 00 00 00 ← CompositionTime 00 00 0A B2 65 88 84 ... ← AVCC 数据长度前缀服务端处理if (avcType 1) { QByteArray avccData pkt-body.mid(5); // ★ AVCC → AnnexB QByteArray annexB avccToAnnexB(avccData, _pubVideoTrack-naluLengthSize()); bool key (frameType 1); auto frame FrameFactory::create( CodecId::H264, annexB, dts, dts, key, false, 0); _publishSource-writeFrame(frame, key); // ★ 写入 MediaSource }6.7 客户端 → 服务器Audio Sequence Header04 00 00 00 00 00 04 08 01 00 00 00 AF ← Audio Tag Header 00 ← AAC sequence header 12 10 ← AudioSpecificConfig服务端处理void RtmpSession::handlePublishAudio(const RtmpPacket::Ptr pkt) { const uint8_t* p (const uint8_t*)pkt-body.constData(); uint8_t soundFormat (p[0] 4) 0x0F; // 10 AAC uint8_t aacType p[1]; // 0 Sequence Header if (soundFormat ! 10) return; if (_pubBaseDts 0) _pubBaseDts int64_t(pkt-timeStamp); uint64_t dts pkt-timeStamp - _pubBaseDts; if (aacType 0) { QByteArray asc pkt-body.mid(2); if (!_pubAudioTrack) { _pubAudioTrack std::make_sharedAudioTrack(CodecId::AAC, 1); // ★ 从 ASC 解析采样率/声道 auto info AdtsUtils::parseAudioSpecificConfig(asc); _pubAudioTrack-setSampleRate(info.sampleRate); _pubAudioTrack-setChannels(info.channelConfig); _pubAudioTrack-setExtradata(asc); _publishSource-addTrack(_pubAudioTrack); } auto frame FrameFactory::create( CodecId::AAC, asc, 0, 0, true, true, 1); _publishSource-writeFrame(frame, false); finalizePublishSource(); return; } }6.8 客户端 → 服务器裸 AAC04 00 03 E8 00 01 84 08 01 00 00 00 AF ← Audio Tag Header 01 ← raw AAC 21 09 0E 2C ... ← 裸 AAC 数据服务端处理if (aacType 1) { QByteArray rawAac pkt-body.mid(2); auto frame FrameFactory::create( CodecId::AAC, rawAac, dts, dts, true, false, 1); _publishSource-writeFrame(frame, true); // ★ 写入 MediaSource }6.9 注册 MediaSource关键void RtmpSession::finalizePublishSource() { auto tracks _publishSource-tracks(); for (auto t : tracks) { if (!t-ready()) return; // 有 track 未就绪等 } if (_publishRegistered) return; _publishSource-regist(); // ★ 注册到 MediaSourceManager _publishRegistered true; qDebug() [RTMP] MediaSource registered: _publishSource-url(); }注册后StreamManager::onTick()每 500ms → 发现新流 rtmp://127.0.0.1/live/test → 创建 RtspMuxer RtmpMuxer → muxer-start() 订阅 RingBuffer → 之后任何协议拉流都能拿到帧七、★ 协议桥接核心机制7.1 桥接原理关键MediaSource的 key 是vhost/app/stream不含 schema。MediaTuple::url()cppQString url() const { return QString(%1://%2/%3/%4).arg(schema).arg(vhost).arg(app).arg(stream); }但MediaSourceManager查找时的 key 只按完整 URL所以// RtspSession 拉流时客户端访问 rtsp://127.0.0.1/live/test MediaTuple rtspTuple { rtsp, 127.0.0.1, live, test }; // 查找失败 // 再换 rtmp 前缀找 MediaTuple rtmpTuple { rtmp, 127.0.0.1, live, test }; // 找到因为推流时注册的就是这个 URL7.2 跨协议查找代码RtmpSession 里的跨协议查找拉流时找 RTSP 推的流_playSource MediaSourceManager::instance().find(tuple); // 先按 rtmp 找 if (!_playSource) { MediaTuple rtspTuple tuple; rtspTuple.schema rtsp; // 换 rtsp 前缀 _playSource MediaSourceManager::instance().find(rtspTuple); }RtspSession 里同理拉流时找 RTMP 推的流_liveSource MediaSourceManager::instance().find(tuple); // 先按 rtsp 找 if (!_liveSource) { MediaTuple rtmpTuple tuple; rtmpTuple.schema rtmp; // 换 rtmp 前缀 _liveSource MediaSourceManager::instance().find(rtmpTuple); }7.3 四种组合都能通推流拉流命令RTMPRTMPffplay rtmp://127.0.0.1:1935/live/testRTMPRTSPffplay -rtsp_transport tcp rtsp://127.0.0.1:554/live/testRTSPRTMPffplay rtmp://127.0.0.1:1935/live/testRTSPRTSPffplay -rtsp_transport tcp rtsp://127.0.0.1:554/live/test这就是一个完整的流媒体服务器——推送协议和拉流协议完全解耦。八、完整时序图8.1 推流方向客户端 服务器 │ │ │──── TCP 连接 ───────────────────────►│ │──── C0C1 (11536B) ────────────────►│ │◄─── S0S1S2 (3073B) ─────────────────│ │──── C2 ( S1, 1536B) ───────────────►│ │ │ │──── connect live ─────────────────►│ RtmpSession::handleConnect │◄─── _result (Success) ────────────────│ → sendConnectResult │◄─── Window Ack Size (5000000) ────────│ │ │ │──── createStream ───────────────────►│ handleCreateStream │◄─── _result (streamId1) ────────────│ │ │ │──── publish test live ──────────►│ handlePublish │◄─── onStatus (Publish.Start) ─────────│ → _publishSource new MediaSource │ │ │──── setDataFrame/onMetaData ───────►│ handleData缓存元数据 │──── Video Sequence Header ──────────►│ handlePublishVideo(avcType0) │ │ → avcConfigToAnnexB → VideoTrack │──── Audio Sequence Header ──────────►│ handlePublishAudio(aacType0) │ │ → parseASC → AudioTrack │ │ → finalizePublishSource → regist() │ │ → StreamManager 创建 RtspMuxer │ │ │──── Video Frame (AVCC) ─────────────►│ handlePublishVideo(avcType1) │ │ → avccToAnnexB → Frame │──── Audio Frame (raw AAC) ──────────►│ → writeFrame → RingBuffer │──── ... 持续 ... │ → RtspMuxer 分发8.2 拉流方向客户端 服务器 │ │ │──── C0C1 → S0S1S2 → C2 ──────────►│ │──── connect ────────────────────────►│ │◄─── _result ──────────────────────────│ │──── createStream ───────────────────►│ │◄─── _result (streamId1) ────────────│ │ │ │──── play test -2000 ──────────────►│ handlePlay │ │ → 跨协议 find MediaSource │ │ → 拿/建 RtmpMuxer │ │ → addSession │◄─── onStatus (Play.Reset) ────────────│ │◄─── onStatus (Play.Start) ────────────│ │◄─── onStatus (Data.Start) ────────────│ │◄─── onMetaData ──────────────────────│ → sendOnMetaData │◄─── Video Sequence Header (avcC) ────│ → sendVideoConfig │◄─── Audio Sequence Header (ASC) ─────│ → sendAudioConfig │ │ │◄─── Video Frame (AVCC) ──────────────│ ← RtmpMuxer 从 RingBuffer 拿帧 │◄─── Audio Frame (raw AAC) ───────────│ ← 转 RTMP 格式发给客户端 │◄─── ... 持续 ... │九、常见问题速查现象原因修复握手后立即断开C0 版本号不是 0x03客户端要用 0x03_result收到但无法继续txn不匹配检查_connectTxn保存是否正确play 后无数据跨协议查找失败检查 URL 的 app/stream 是否一致publish 后拉流 404MediaSource 未 regist检查所有 track 是否 ready推流成功但花屏avcC 解析错检查avcConfigToAnnexB的字段偏移音频无声ASC 解析错从 ASC 解析采样率不要用 tag bit拉流断流Muxer 未正确 addSession检查_rtmpMuxer-addSession(this)时间戳跳变未做 DTS 基准归一化用_pubBaseDts归一化到 0十、小结RTMP 服务端的核心就四件事1. 握手接收方视角收到 C0C1回 S0S1S2消费 C2不校验关键S2 C1 回显2. 命令处理connect→ 保存 app回_resultcreateStream→ 分配 streamId生产要自增play→ 跨协议查找 Muxer 分发publish→ 创建 MediaSource3. 推流方向AVCC ↔ AnnexB收到avcC→ 转AnnexB→ 存入Track::extradata收到AVCC→ 转AnnexB→ 写入MediaSource收到ASC→ 解析采样率/声道4. 拉流方向AnnexB → AVCCTrack::extradataAnnexB→ 拼AVCDecoderConfigurationRecord→ 发送每一帧AnnexB→ 转AVCC→ 拼 RTMP video tag → 发送三条核心边界边界接口协议 ↔ 数据MediaSource::writeFrame()/regist()推流 ↔ 拉流MediaSourceManager::find()跨协议查Muxer ↔ 客户端Muxer::addSession()/onTrackFrame()把这三条理清一个支持 RTSP/RTMP 双向桥接的流媒体服务器就立起来了。任何协议推上来的流都能被任何协议拉走——这就是协议桥接的价值。rtsp,rtmp