深入浅出SRS—RTMP实现

2024-09-07 02:28
文章标签 实现 深入浅出 rtmp srs

本文主要是介绍深入浅出SRS—RTMP实现,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

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 连接都是相同的处理流程。

相关源码如下:

if (!resource) {if (listener == rtmp_listener_) {resource = new SrsRtmpConn(this, stfd2, ip, port);}...
}// 添加到容器进行集中管理
conn_manager->add(resource);// 启动连接的事件循环,开始处理报文
ISrsStartable* conn = dynamic_cast<ISrsStartable*>(resource);
if ((err = conn->start()) != srs_success) {return srs_error_wrap(err, "start conn coroutine");
}

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 直播实现所需的媒体处理。

这篇关于深入浅出SRS—RTMP实现的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1143795

相关文章

hdu1043(八数码问题,广搜 + hash(实现状态压缩) )

利用康拓展开将一个排列映射成一个自然数,然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu

【C++】_list常用方法解析及模拟实现

相信自己的力量,只要对自己始终保持信心,尽自己最大努力去完成任何事,就算事情最终结果是失败了,努力了也不留遗憾。💓💓💓 目录   ✨说在前面 🍋知识点一:什么是list? •🌰1.list的定义 •🌰2.list的基本特性 •🌰3.常用接口介绍 🍋知识点二:list常用接口 •🌰1.默认成员函数 🔥构造函数(⭐) 🔥析构函数 •🌰2.list对象

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯,先赞后看哦~🎈🎈 🏆 作者简介:景天科技苑 🏆《头衔》:大厂架构师,华为云开发者社区专家博主,阿里云开发者社区专家博主,CSDN全栈领域优质创作者,掘金优秀博主,51CTO博客专家等。 🏆《博客》:Python全栈,前后端开发,小程序开发,人工智能,js逆向,App逆向,网络系统安全,数据分析,Django,fastapi

让树莓派智能语音助手实现定时提醒功能

最初的时候是想直接在rasa 的chatbot上实现,因为rasa本身是带有remindschedule模块的。不过经过一番折腾后,忽然发现,chatbot上实现的定时,语音助手不一定会有响应。因为,我目前语音助手的代码设置了长时间无应答会结束对话,这样一来,chatbot定时提醒的触发就不会被语音助手获悉。那怎么让语音助手也具有定时提醒功能呢? 我最后选择的方法是用threading.Time

Android实现任意版本设置默认的锁屏壁纸和桌面壁纸(两张壁纸可不一致)

客户有些需求需要设置默认壁纸和锁屏壁纸  在默认情况下 这两个壁纸是相同的  如果需要默认的锁屏壁纸和桌面壁纸不一样 需要额外修改 Android13实现 替换默认桌面壁纸: 将图片文件替换frameworks/base/core/res/res/drawable-nodpi/default_wallpaper.*  (注意不能是bmp格式) 替换默认锁屏壁纸: 将图片资源放入vendo

C#实战|大乐透选号器[6]:实现实时显示已选择的红蓝球数量

哈喽,你好啊,我是雷工。 关于大乐透选号器在前面已经记录了5篇笔记,这是第6篇; 接下来实现实时显示当前选中红球数量,蓝球数量; 以下为练习笔记。 01 效果演示 当选择和取消选择红球或蓝球时,在对应的位置显示实时已选择的红球、蓝球的数量; 02 标签名称 分别设置Label标签名称为:lblRedCount、lblBlueCount

Kubernetes PodSecurityPolicy:PSP能实现的5种主要安全策略

Kubernetes PodSecurityPolicy:PSP能实现的5种主要安全策略 1. 特权模式限制2. 宿主机资源隔离3. 用户和组管理4. 权限提升控制5. SELinux配置 💖The Begin💖点点关注,收藏不迷路💖 Kubernetes的PodSecurityPolicy(PSP)是一个关键的安全特性,它在Pod创建之前实施安全策略,确保P

工厂ERP管理系统实现源码(JAVA)

工厂进销存管理系统是一个集采购管理、仓库管理、生产管理和销售管理于一体的综合解决方案。该系统旨在帮助企业优化流程、提高效率、降低成本,并实时掌握各环节的运营状况。 在采购管理方面,系统能够处理采购订单、供应商管理和采购入库等流程,确保采购过程的透明和高效。仓库管理方面,实现库存的精准管理,包括入库、出库、盘点等操作,确保库存数据的准确性和实时性。 生产管理模块则涵盖了生产计划制定、物料需求计划、

C++——stack、queue的实现及deque的介绍

目录 1.stack与queue的实现 1.1stack的实现  1.2 queue的实现 2.重温vector、list、stack、queue的介绍 2.1 STL标准库中stack和queue的底层结构  3.deque的简单介绍 3.1为什么选择deque作为stack和queue的底层默认容器  3.2 STL中对stack与queue的模拟实现 ①stack模拟实现

基于51单片机的自动转向修复系统的设计与实现

文章目录 前言资料获取设计介绍功能介绍设计清单具体实现截图参考文献设计获取 前言 💗博主介绍:✌全网粉丝10W+,CSDN特邀作者、博客专家、CSDN新星计划导师,一名热衷于单片机技术探索与分享的博主、专注于 精通51/STM32/MSP430/AVR等单片机设计 主要对象是咱们电子相关专业的大学生,希望您们都共创辉煌!✌💗 👇🏻 精彩专栏 推荐订阅👇🏻 单片机