当前位置：首页 > news >正文

深入浅出SRS—RTMP实现

news 2025/7/13 7:34:52

RTMP 直播是 SRS 最典型的使用场景，客户端使用 RTMP 协议向 SRS 推流，使用 RTMP 协议从 SRS 拉流，SRS 作为一个 RTMP 直播服务器实现媒体的转发。同时，RTMP 是 SRS 的中转协议，其他协议之间的互通需要先转为 RTMP，因此，理解 SRS RTMP 直播实现是理解其他协议实现的重要前提。本文主要分析 SRS RTMP 直播功能的实现原理，相关概念和配置请参考官方文档。

1. Origin

单机版 SRS，加上一个推流端和一个拉流端，组成最简单的 RTMP 直播场景，如下图所示。其中 RtmpConn 表示一个 RTMP 连接，LiveSouce 表示一个直播资源，接下来我们来分析这个直播场景的实现。

1.1. 创建连接

在初始化阶段，SrsServer 会读取配置文件使用 SrsMultipleTcpListeners 建立监听。SrsMultipleTcpListeners 把自己装扮成一个 ISrsListener，但其内部包含的 SrsTcpListener 才是真正的监听者，支持多个监听者。

SrsTcpLisener 启动后，当有客户端连接上来，SrsTcpLisener 会通知 SrsMultipleTcpListeners，SrsMultipleTcpListeners 再通知 SrsServer，SrsServer 创建一个 SrsRtmpConn，添加到 SrsResourceManger（用来管理所有 RTMP 连接），并启动 SrsRtmpConn 事件循环。Publisher 和 palyer 连接都是相同的处理流程。

1.2. 协议交互

连接建立成功后，下一步就要按照 RTMP 协议规范，完成 handshake、connect 等协议交互，当然还会有其他参数设置。

在 SrsRtmpConn 的事件循环中，可以看到 handshake 和 connect 动作。因为 RTMP 协议是 CS 模式，不管是推流还是拉流，都是客户端主动发起连接，然后执行收发媒体之前的协议交互，处理逻辑是一样的。

srs_error_t SrsRtmpConn::do_cycle()
{...// 完成 handshakeif ((err = rtmp->handshake()) != srs_success) {return srs_error_wrap(err, "rtmp handshake");}...// 完成 connectSrsRequest* req = info->req;if ((err = rtmp->connect_app(req)) != srs_success) {return srs_error_wrap(err, "rtmp connect tcUrl");}...// 进入接收媒体数据循环if ((err = service_cycle()) != srs_success) {err = srs_error_wrap(err, "service cycle");}...
}

1.3. 识别类型

在处理媒体数据之前，需要能识别客户端是 publisher 还是 player。SRS 根据具体协议定义了一系列连接类型，分为 publishers 和 players 两大类。我们当前分析的场景，只需要关注 SrsRtmpConnPlay 和 SrsRtmpConnFMLEPublish 两种类型即可。

enum SrsRtmpConnType
{SrsRtmpConnUnknown = 0x0000,// All players.SrsRtmpConnPlay = 0x0100,SrsHlsPlay = 0x0101,SrsFlvPlay = 0x0102,SrsRtcConnPlay = 0x0110,SrsSrtConnPlay = 0x0120,// All publishers.SrsRtmpConnFMLEPublish = 0x0200,SrsRtmpConnFlashPublish = 0x0201,SrsRtmpConnHaivisionPublish = 0x0202,SrsRtcConnPublish = 0x0210,SrsSrtConnPublish = 0x0220,
};

SRS 先调用 identify_client 方法识别客户端类型，然后根据客户端类型确定是做推流处理还是拉流处理。

srs_error_t SrsRtmpConn::stream_service_cycle()
{// 解析客户端类型if ((err = rtmp->identify_client(info->res->stream_id, info->type, req->stream, req->duration)) != srs_success) {return srs_error_wrap(err, "rtmp: identify client");}...// 根据客户端类型进行对应处理switch (info->type) {case SrsRtmpConnPlay: {...return playing(source); // 拉流}case SrsRtmpConnFMLEPublish: {...return publishing(source); // 推流}...}...
}

identify_client 根据收到的消息来判定客户端类型，推流端才会发送以下消息：

#define RTMP_AMF0_COMMAND_RELEASE_STREAM        "releaseStream"
#define RTMP_AMF0_COMMAND_FC_PUBLISH            "FCPublish"
#define RTMP_AMF0_COMMAND_UNPUBLISH             "FCUnpublish"

拉流端才会发送以下消息：

#define RTMP_AMF0_COMMAND_PLAY                  "play"

srs_error_t SrsRtmpServer::identify_client(int stream_id, SrsRtmpConnType& type, string& stream_name, srs_utime_t& duration)
{    while (true) {SrsCommonMessage* msg = NULL;if ((err = protocol->recv_message(&msg)) != srs_success) {return srs_error_wrap(err, "recv identify message");}...if (dynamic_cast<SrsCreateStreamPacket*>(pkt)) {// 内部做进一步分析return identify_create_stream_client(dynamic_cast<SrsCreateStreamPacket*>(pkt), stream_id, 3, type,stream_name, duration);}if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {// type = SrsRtmpConnFMLEPublishreturn identify_fmle_publish_client(dynamic_cast<SrsFMLEStartPacket*>(pkt), type, stream_name);}if (dynamic_cast<SrsPlayPacket*>(pkt)) {// type = SrsRtmpConnPlayreturn identify_play_client(dynamic_cast<SrsPlayPacket*>(pkt), type, stream_name, duration);}}
}

1.4. Publish

如果识别是连接的客户端是 publisher，进入 publishing 处理流程。

1.4.1. 准备工作

推流端发送 publish 消息，SrsRtmpConn 会创建一个 SrsLiveSource，SrsLiveSource 是直播源的逻辑实体，每个推流在 SRS 上都会有一个 SrsLiveSource 对应。

srs_error_t SrsRtmpConn::stream_service_cycle()
{...// 可能需要创建 SrsLiveSourceSrsLiveSource* source = NULL;if ((err = _srs_sources->fetch_or_create(req, server, &source)) != srs_success) {return srs_error_wrap(err, "rtmp: fetch source");}switch (info->type) {...case SrsRtmpConnFMLEPublish: {if ((err = rtmp->start_fmle_publish(info->res->stream_id)) != srs_success) {return srs_error_wrap(err, "rtmp: start FMLE publish");}return publishing(source);}...}return err;
}

可能是基于性能考量，SRS 创建了一个新的协程来专门接收推流媒体数据，一切准备妥当，通知客户端可以发送媒体数据了，进入 SrsRtmpConn 的事件循环。

srs_error_t SrsRtmpConn::do_publishing(SrsLiveSource* source, SrsPublishRecvThread* rtrd)
{srs_error_t err = srs_success;SrsRequest* req = info->req;SrsPithyPrint* pprint = SrsPithyPrint::create_rtmp_publish();SrsAutoFree(SrsPithyPrint, pprint);// 这里面启动了一个新的协程接收媒体数据if ((err = rtrd->start()) != srs_success) {return srs_error_wrap(err, "rtmp: receive thread");}...// 发送 NetStream.Publish.Start 消息，通知客户端开始发送媒体数据if ((err = rtmp->start_publishing(info->res->stream_id)) != srs_success) {return srs_error_wrap(err, "start publishing");}while (true) {if ((err = trd->pull()) != srs_success) {return srs_error_wrap(err, "rtmp: thread quit");}...}return err;
}

1.4.2. 接收媒体

刚刚讲到，SRS 启动一个新的协程来接收媒体数据，这个协程就是 SrsRecvThread，只是它被 SrsPublishRecvThread 包裹了一层。SrsRecvThread 启动后，使用 SrsRtmpServer 从网络接收媒体数据，收到的媒体数据经过 SrsPublishRecvThread 传递到 SrsRtmpConn。

具体来说，SrsRecvThread 调用 SrsRtmpServer::recv_message 从网络读取媒体数据，并调用 consume 接口发送到 SrsPublishRecvThread。

srs_error_t SrsRecvThread::do_cycle()
{    while (true) {...// 这里的 pumper 就是 SrsPublishRecvThreadSrsCommonMessage* msg = NULL;if ((err = rtmp->recv_message(&msg)) == srs_success) {err = pumper->consume(msg);}...}return err;
}

SrsPublishRecvThread 只是一个二道贩子，继续把数据发送到 SrsRtmpConn 处理。

srs_error_t SrsPublishRecvThread::consume(SrsCommonMessage* msg)
{...// the rtmp connection will handle this messageerr = _conn->handle_publish_message(_source, msg);...
}

SrsRtmpConn 根据协议头判断是音频数据还是视频数据，然后将数据交给 SrsLiveSource 处理。

srs_error_t SrsRtmpConn::process_publish_message(SrsLiveSource* source, SrsCommonMessage* msg)
{...// process audio packetif (msg->header.is_audio()) {if ((err = source->on_audio(msg)) != srs_success) {return srs_error_wrap(err, "rtmp: consume audio");}return err;}// process video packetif (msg->header.is_video()) {if ((err = source->on_video(msg)) != srs_success) {return srs_error_wrap(err, "rtmp: consume video");}return err;}...
}

1.4.3. 分发媒体

SrsLiveSource 收到媒体数据后（以视频数据为例），也没做太多处理，主要是进行数据分发，分发对象有三个：

1）SrsLiveConsumer：RTMP 直播观看者。

2）SrsRtcFromRtmpBridge：RTMP 转 RTC。

3）SrsOriginHub：DVR（录制）、HLS（HLS 直播观看）、DASH（DASH 直播观看）等。

srs_error_t SrsLiveSource::on_video_imp(SrsSharedPtrMessage* msg)
{...// 解析媒体数据if ((err = format_->on_video(msg)) != srs_success) {return srs_error_wrap(err, "format consume video");}...// HUB 转发if ((err = hub->on_video(msg, is_sequence_header)) != srs_success) {return srs_error_wrap(err, "hub consume video");}// 桥接转发if (bridge_ && (err = bridge_->on_video(msg)) != srs_success) {return srs_error_wrap(err, "bridge consume video");}// RTMP player 转发if (!drop_for_reduce) {for (int i = 0; i < (int)consumers.size(); i++) {SrsLiveConsumer* consumer = consumers.at(i);if ((err = consumer->enqueue(msg, atc, jitter_algorithm)) != srs_success) {return srs_error_wrap(err, "consume video");}}}...
}

SrsOriginHub 内部将媒体数据转发给 HLS、DASH、DVR 和 fowarder（Oring 热备）。

srs_error_t SrsOriginHub::on_video(SrsSharedPtrMessage* shared_video, bool is_sequence_header)
{...// HLSif ((err = hls->on_video(msg, format)) != srs_success) {std::string hls_error_strategy = _srs_config->get_hls_on_error(req_->vhost);if (srs_config_hls_is_on_error_ignore(hls_error_strategy)) {srs_warn("hls: ignore video error %s", srs_error_desc(err).c_str());hls->on_unpublish();srs_error_reset(err);} else if (srs_config_hls_is_on_error_continue(hls_error_strategy)) {if (srs_hls_can_continue(srs_error_code(err), source->meta->vsh(), msg)) {srs_error_reset(err);} else {return srs_error_wrap(err, "hls: video");}} else {return srs_error_wrap(err, "hls: video");}}// DASHif ((err = dash->on_video(msg, format)) != srs_success) {srs_warn("dash: ignore video error %s", srs_error_desc(err).c_str());srs_error_reset(err);dash->on_unpublish();}// DVRif ((err = dvr->on_video(msg, format)) != srs_success) {srs_warn("dvr: ignore video error %s", srs_error_desc(err).c_str());srs_error_reset(err);dvr->on_unpublish();}// Forwarderif (!forwarders.empty()) {std::vector<SrsForwarder*>::iterator it;for (it = forwarders.begin(); it != forwarders.end(); ++it) {SrsForwarder* forwarder = *it;if ((err = forwarder->on_video(msg)) != srs_success) {return srs_error_wrap(err, "forward video");}}}return err;
}

1.5. Play

如果识别是连接的客户端是 publisher，进入 playing 处理流程。

1.5.1. 准备工作

播放之前需要先查找直播源，如果直播源不存在，会触发提前创建一个 SrsLiveSource，等待后面直播源连上来后发送媒体数据。

srs_error_t SrsRtmpConn::stream_service_cycle()
{...// 查找直播源（直播源没连上也没关系，会触发提前创建 SrsLiveSource）SrsLiveSource* source = NULL;if ((err = _srs_sources->fetch_or_create(req, server, &source)) != srs_success) {return srs_error_wrap(err, "rtmp: fetch source");}switch (info->type) {case SrsRtmpConnPlay: {// 响应客户端的 play 请求if ((err = rtmp->start_play(info->res->stream_id)) != srs_success) {return srs_error_wrap(err, "rtmp: start play");}// 进入播放处理return playing(source);}...}return err;
}

每一个 RTMP player，都会创建一个 SrsLiveConsumer，SrsLiveConsumer 是 RTMP player 在 SRS 上的逻辑实体。和 publishing 一样，SRS 会启动一个新的协程从客户端接收消息。

srs_error_t SrsRtmpConn::playing(SrsLiveSource* source)
{...// 创建一个 consumerSrsLiveConsumer* consumer = NULL;SrsAutoFree(SrsLiveConsumer, consumer);if ((err = source->create_consumer(consumer)) != srs_success) {return srs_error_wrap(err, "rtmp: create consumer");}// Use receiving thread to receive packets from peer.SrsQueueRecvThread trd(consumer, rtmp, SRS_PERF_MW_SLEEP, _srs_context->get_id());// 启动新的协程从客户端收取消息if ((err = trd.start()) != srs_success) {return srs_error_wrap(err, "rtmp: start receive thread");}err = do_playing(source, consumer, &trd);...
}

1.5.2. 接收消息

SrsRecvThread 通过 SrsRtmpServer 从 player 接收控制消息，比如暂停、恢复等。收到的消息会放到 SrsQueueRecvThread 内部的一个队列中，SrsRtmpConn 在事件循环中从 SrsQueueRecvThread 的队列拉取消息，作相应处理，然后通过 SrsRtmpServer 向 player 发送响应。

收到的控制消息插入 SrsQueueRecvThread 队列。

srs_error_t SrsQueueRecvThread::consume(SrsCommonMessage* msg)
{queue.push_back(msg);return srs_success;
}

SrsRtmpConn 在事件循环中拉取消息并进行处理。

srs_error_t SrsRtmpConn::do_playing(SrsLiveSource* source, SrsLiveConsumer* consumer, SrsQueueRecvThread* rtrd)
{...while (true) {...// 从客户端收到的消息都放在队列中，通过 pump 方法从队列中获取收到的消息while (!rtrd->empty()) {SrsCommonMessage* msg = rtrd->pump(); // SrsQueueRecvThread::pumpif ((err = process_play_control_msg(consumer, msg)) != srs_success) {return srs_error_wrap(err, "rtmp: play control message");}}...}return err;
}

1.5.3. 发送媒体

前面 publishing 讲过，Publiser 的媒体数据会分发到 SrsLiveConsumer。Player 的 SrsRtmpConn 事件循环从 SrsLiveConsumer 队列中拉取媒体数据，通过 SrsRtmpServer 发送到客户端。

srs_error_t SrsRtmpConn::do_playing(SrsLiveSource* source, SrsLiveConsumer* consumer, SrsQueueRecvThread* rtrd)
{...while (true) {...// 拉取媒体数据int count = (send_min_interval > 0)? 1 : 0;if ((err = consumer->dump_packets(&msgs, count)) != srs_success) {return srs_error_wrap(err, "rtmp: consumer dump packets");}...// 发送媒体数据if (count > 0 && (err = rtmp->send_and_free_messages(msgs.msgs, count, info->res->stream_id)) != srs_success) {return srs_error_wrap(err, "rtmp: send %d messages", count);}...}return err;
}

2. Edge cluster

当直播客户端越来越多，Origin 已经扛不住这么大的并发和流量，必须上集群，Edge cluster 就是这样一种集群，用来解决大量直播观众带来的挑战与问题。

2.1. Play

当某个直播源的直播观众非常多的时候，可以考虑引入 Edge cluster。流还是推到 Origin，拉流则从 Edge cluster 拉，如下图所示。

Origin 与之前保持一致，因为对于 Origin 来说，Edge 就相当于一个 player。Edge 与 Origin 相比有一些变化，变化主要在推流侧。对应 Origin 的 Client 推流，Edge 其实是使用 Ingester 从 Origin 拉流，支持使用 RMTP 和 FLV 两种拉流方式。

直播源的第一个 player 来拉流的时候，会触发 Edge 从 Origin 拉流。

srs_error_t SrsLiveSource::create_consumer(SrsLiveConsumer*& consumer)
{srs_error_t err = srs_success;consumer = new SrsLiveConsumer(this);consumers.push_back(consumer);// There should be one consumer, so reset the timeout.stream_die_at_ = 0;publisher_idle_at_ = 0;// 如果是 edge，通知 SrsPlayEdge 从 oring 拉流if (_srs_config->get_vhost_is_edge(req->vhost)) {if ((err = play_edge->on_client_play()) != srs_success) {return srs_error_wrap(err, "play edge");}}return err;
}

SrsEdgeIngester 内部使用 SrsBasicRtmpClient 模拟一个 RTMP player 从 Origin 拉流，拉到的媒体数据回调到 SrsLiveSource，后面的分发逻辑与 Origin 保持一致。

2.2. Publish

Edge 既可以从 Origin 拉流，也可以推流到 Origin，如下图所示。至于为什么不直接推 Origin，可能是 Origin 不可达，或者 Client 到 Edge 的网络比到 Origin 好。

处理逻辑分为三部分，如下图所示（使用不同的颜色区分）。第一部分是调用 SrsRtmpConn::acquire_publish 与 Origin 建立推送通道；第二部分是收到媒体数据插入到 SrsEdgeForwarder 的消息队列；第三部分是 SrsEdgeForwrader 事件循环从消息队列中拉取媒体数据发送到 Origin。

SRS 在处理 publish 时，Edge 和 Origin 走不同的分支。

srs_error_t SrsRtmpConn::acquire_publish(SrsLiveSource* source)
{...if (info->edge) {// Edge 处理分支err = source->on_edge_start_publish();} else {// Origin 处理分支err = source->on_publish();}return err;
}

如果是 Edge，则会启动 SrsEdgeForwarder 推流到 Origin。

srs_error_t SrsPublishEdge::on_client_publish()
{...// 启动 forwarder 推流到 originerr = forwarder->start();...
}

收到媒体数据时，直接转发到 origin。

srs_error_t SrsRtmpConn::process_publish_message(SrsLiveSource* source, SrsCommonMessage* msg)
{srs_error_t err = srs_success;// 如果是 edge，媒体数据直接转发到 originif (info->edge) {if ((err = source->on_edge_proxy_publish(msg)) != srs_success) {return srs_error_wrap(err, "rtmp: proxy publish");}return err;}...
}

3. Origin cluster

当直播源不多的时候，使用一个 Origin 就行（不考虑高可用），此时不存在负载均衡和回哪个源的问题。如果直播源越来越多，一个 Origin 承受不起并发和流量时，此时就需要使用 Origin cluster。但这也带来一个问题，直播源分布在多个 Origin 上，该如何调度直播观看端或者 Edge 到哪个 Origin 拉流。Origin Cluster 通过开放集群内资源查询接口，实现基于 RTMP302 的集群资源自动调度。

如果调度到没有此直播源的 Origin，则 Client 可能等到海枯石烂也徒劳，如下图所示。

Origin Cluster 通过开放集群内资源查询接口，实现基于 RTMP302 的集群资源自动调度。

如果本地没有直播源，SrsLiveSource 状态为 inactive。如果没有配置 Origin cluster，SRS 会等待直播源连上来；如果配置了 Origin cluster，则会向集群中其他 Origin 发起资源查询，

srs_error_t SrsRtmpConn::playing(SrsLiveSource* source)
{...// 如果本服务是 origin，且配置了 origin 集群，且在本服务没有找到直播源if (!info->edge && // Origin_srs_config->get_vhost_origin_cluster(req->vhost) &&  // Origin clustersource->inactive()) {// 获取 origin 集群的地址vector<string> coworkers = _srs_config->get_vhost_coworkers(req->vhost);// 遍历集群中所有 origin 进行查询for (int i = 0; i < (int)coworkers.size(); i++) {string host; int port = 0; string coworker = coworkers.at(i);// 构造资源查询 URLstring url = "http://" + coworker + "/api/v1/clusters?"+ "vhost=" + req->vhost + "&ip=" + req->host + "&app=" + req->app + "&stream=" + req->stream+ "&coworker=" + coworker;// 发起资源查询请求if ((err = SrsHttpHooks::discover_co_workers(url, host, port)) != srs_success) {// 向所有 origin 发起查询if (i < (int)coworkers.size() - 1) {continue;}// 都查不到返回失败return srs_error_wrap(err, "discover coworkers, url=%s", url.c_str());}// 构造 RTMP302string rurl = srs_generate_rtmp_url(host, port, req->host, req->vhost, req->app, req->stream, req->param);// 向客户端或 edge 发送 RTMP302bool accepted = false;if ((err = rtmp->redirect(req, rurl, accepted)) != srs_success) {srs_error_reset(err);} else {// 重定向完成后关闭连接return srs_error_new(ERROR_CONTROL_REDIRECT, "redirected");}}return srs_error_new(ERROR_OCLUSTER_REDIRECT, "no origin");}...
}

4. Forward

大部分场景只需要按需拉流逻辑，但也有些场景需要服务器推流，比如热备、小型集群等场景，Forward 特性补全了 SRS 的推流能力。

如下所示，中间配置了 Forward 的 Origin 可以将直播流通过 Forwarder 推给其他 Origin，没有配置 Forward 的 Origin 感知不到任何变化，从 Origin 推过来的流与客户端推过来的流是一样的处理逻辑。

当有直播源开始 publish 时，SRS 会尝试创建针对此直播源的 forwarder。

srs_error_t SrsOriginHub::on_publish()
{srs_error_t err = srs_success;// create forwardersif ((err = create_forwarders()) != srs_success) {return srs_error_wrap(err, "create forwarders");}...
}

SrsOriginHub 读取配置创建 forwarder，SRS 支持静态和动态两种 forwarder，可以根据需求选择。所谓动态 forwarder，就是 SRS 会通过 Hook 回调应用层，应用层返回转发目的地址。所谓静态 forwarder，就是配置文件中配置的目的地址。

srs_error_t SrsOriginHub::create_forwarders()
{srs_error_t err = srs_success;// 未开启 forwarder，返回if (!_srs_config->get_forward_enabled(req_->vhost)) {return err;}// 动态 forward 配置bool applied_backend_server = false;if ((err = create_backend_forwarders(applied_backend_server)) != srs_success) {return srs_error_wrap(err, "create backend applied=%d", applied_backend_server);}// 动态 forward 优先，配置了动态 forward，则忽略静态 forwardif (applied_backend_server) {return err;}// 静态 forward 配置SrsConfDirective* conf = _srs_config->get_forwards(req_->vhost);// 遍历 destination 配置的所有地址for (int i = 0; conf && i < (int)conf->args.size(); i++) {std::string forward_server = conf->args.at(i);// 创建 forwarderSrsForwarder* forwarder = new SrsForwarder(this);forwarders.push_back(forwarder);// 初始化 forwarderif ((err = forwarder->initialize(req_, forward_server)) != srs_success) {return srs_error_wrap(err, "init forwarder");}srs_utime_t queue_size = _srs_config->get_queue_length(req_->vhost);forwarder->set_queue_size(queue_size);if ((err = forwarder->on_publish()) != srs_success) {return srs_error_wrap(err, "start forwarder failed, vhost=%s, app=%s, stream=%s, forward-to=%s",req_->vhost.c_str(), req_->app.c_str(), req_->stream.c_str(), forward_server.c_str());}}return err;
}

收到直播源的媒体数据后，SRS 会将媒体数据转发给直播源的所有 fowarder。

srs_error_t SrsOriginHub::on_video(SrsSharedPtrMessage* shared_video, bool is_sequence_header)
{...// copy to all forwarders.if (!forwarders.empty()) {std::vector<SrsForwarder*>::iterator it;for (it = forwarders.begin(); it != forwarders.end(); ++it) {SrsForwarder* forwarder = *it;if ((err = forwarder->on_video(msg)) != srs_success) {return srs_error_wrap(err, "forward video");}}}return err;
}

5. 媒体处理

5.1. Seuqence header

RTMP 协议的媒体数据使用 FLV 封装，编解码器一般选择 AVC/AAC 组合。

对于 AVC 来说，会先传输 AVC sequence header， AVC sequence header 通常包含 AVC 的配置信息，其中包括 SPS（Sequence Parameter Set）和 PPS（Picture Parameter Set）。

对于 AAC 来说，会先传输 AAC sequence header，AAC sequence header 通常包含采样率、采样位深、通道数、AOT（Audio Object Type）等一些元数据。

Sequence header 通常只在连接建立时发送一次，SRS 会解析 sequence header 并将解析结果保存到 SrsFormat，以备后用。

if ((err = format_->on_audio(msg)) != srs_success) {return srs_error_wrap(err, "format consume audio");
}if ((err = format_->on_video(msg)) != srs_success) {return srs_error_wrap(err, "format consume video");
}

同时还会将包含 sequence header 的报文缓存起来。

if (is_sequence_header || !meta->ash()) {if ((err = meta->update_ash(msg)) != srs_success) {return srs_error_wrap(err, "meta consume audio");}
}if (is_sequence_header && (err = meta->update_vsh(msg)) != srs_success) {return srs_error_wrap(err, "meta update video");
}

需要 sequence header 元数据的地方，如果是 RTMP 协议转发，可以直接发送缓存报文；但如果是不同协议的转发，则需要使用 sequence header 的解析结果。

比如，后面加入的 RTMP player，必须要先发送 sequence header，否则 player 无法解码。

srs_error_t SrsRtmpConn::playing(SrsLiveSource* source)
{...// 创建 consumerSrsLiveConsumer* consumer = NULL;SrsAutoFree(SrsLiveConsumer, consumer);if ((err = source->create_consumer(consumer)) != srs_success) {return srs_error_wrap(err, "rtmp: create consumer");}// 立即向 consumer 发送 sequence headerif ((err = source->consumer_dumps(consumer)) != srs_success) {return srs_error_wrap(err, "rtmp: dumps consumer");}...
}

再比如，启动新的 forwrader 时，也需要发送 sequence header。

srs_error_t SrsOriginHub::on_forwarder_start(SrsForwarder* forwarder)
{srs_error_t err = srs_success;SrsSharedPtrMessage* cache_metadata = source->meta->data();SrsSharedPtrMessage* cache_sh_video = source->meta->vsh();SrsSharedPtrMessage* cache_sh_audio = source->meta->ash();// feed the forwarder the metadata/sequence header,// when reload to enable the forwarder.if (cache_metadata && (err = forwarder->on_meta_data(cache_metadata)) != srs_success) {return srs_error_wrap(err, "forward metadata");}if (cache_sh_video && (err = forwarder->on_video(cache_sh_video)) != srs_success) {return srs_error_wrap(err, "forward video sh");}if (cache_sh_audio && (err = forwarder->on_audio(cache_sh_audio)) != srs_success) {return srs_error_wrap(err, "forward audio sh");}return err;
}

5.2. GOP cache

GOP cache 主要用来提升直播观看的体验。如果没有开启 GOP cache，新加入的直播客户端，要等到下一个关键帧才能够播放视频；如果开启了 GOP cache，新加入的直播客户端可以立即播放视频。

GOP cap 的实现逻辑比较简单，每次收到音频或视频消息都调用 SrsGopCache::cache。

srs_error_t SrsLiveSource::on_video_imp(SrsSharedPtrMessage* msg)
{...// cache the last gop packetsif ((err = gop_cache->cache(msg)) != srs_success) {return srs_error_wrap(err, "gop cache consume vdieo");}...
}

SrsGopCache 默认只缓存当前 GOP 的视频帧，同时还会处理最大缓存帧数限制。


srs_error_t SrsGopCache::cache(SrsSharedPtrMessage* shared_msg)
{srs_error_t err = srs_success;// 配置没有开启if (!enable_gop_cache) {return err;}// the gop cache know when to gop it.SrsSharedPtrMessage* msg = shared_msg;// 只缓存 H264if (msg->is_video()) {if (!SrsFlvVideo::h264(msg->payload, msg->size)) {return err;}        cached_video_count++;audio_after_last_video_count = 0;}// 没缓存任何视频帧if (pure_audio()) {return err;}// 统计自缓存上一帧视频以来收到的音频报文数量if (msg->is_audio()) {audio_after_last_video_count++;}// 超过 3 秒没有收到视频帧了，清空视频缓存if (audio_after_last_video_count > SRS_PURE_AUDIO_GUESS_COUNT) {srs_warn("clear gop cache for guess pure audio overflow");clear();return err;}// 收到关键帧刷新缓存，重新开始缓存if (msg->is_video() && SrsFlvVideo::keyframe(msg->payload, msg->size)) {clear();cached_video_count = 1;}// cache the frame.gop_cache.push_back(msg->copy());// 如果超过配置的最大缓存帧数，也要刷新if (gop_cache_max_frames_ > 0 && gop_cache.size() > (size_t)gop_cache_max_frames_) {clear();}return err;
}

新加入的 RTMP 直播客户端，SRS 服务器会在发送 sequence header 后，立即发送 GOP cache。

srs_error_t SrsLiveSource::consumer_dumps(SrsLiveConsumer* consumer, bool ds, bool dm, bool dg)
{...// If stream is publishing, dumps the sequence header and gop cache.if (hub->active()) {// Copy metadata and sequence header to consumer.if ((err = meta->dumps(consumer, atc, jitter_algorithm, dm, ds)) != srs_success) {return srs_error_wrap(err, "meta dumps");}// copy gop cache to client.if (dg && (err = gop_cache->dump(consumer, atc, jitter_algorithm)) != srs_success) {return srs_error_wrap(err, "gop cache dumps");}}return err;
}

5.3. Chunk

RTMP 协议以 message 为处理单位，其负载采用 FLV 封装，每个视频帧被封装为一个 FLV tag。RTMP 基于 chunk 进行传输，chunk size 默认为 128 bytes，可以动态设置。如果 message 长度超过 chunk size，则 message 会被切分为多个 chunk。接收端会将多个 chunk 重组成 message 后再进行处理。

srs_error_t SrsProtocol::recv_message(SrsCommonMessage** pmsg)
{*pmsg = NULL;srs_error_t err = srs_success;while (true) {SrsCommonMessage* msg = NULL;if ((err = recv_interlaced_message(&msg)) != srs_success) {srs_freep(msg);return srs_error_wrap(err, "recv interlaced message");}// 没有收到完整的 message，继续接收if (!msg) {continue;}...}return err;
}

一般来说，FLV 一个 tag 只会封装一个 NALU，如果一个 tag 要封装多个 NALU，则需要使用 AnnexB 格式，插入起始码。

5.4. CompositionTime

FLV 封装格式中，有 Timestamp 和 CompositionTime 两个时间，其中 Timestamp 一般当做视频帧的 DTS，PTS = Timestamp + CompositionTime。CompositonTime 是有符号整数，可以为0、正数或负数。

6. 总结

本文分析了 SRS RTMP 直播的实现，覆盖了单机、Edge 集群、Origin 集群、Forward 等多种场景，详细讲解了单机场景下全流程处理，重点分析了 RTMP 直播实现所需的媒体处理。