深入浅出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

相关文章

Linux下删除乱码文件和目录的实现方式

《Linux下删除乱码文件和目录的实现方式》:本文主要介绍Linux下删除乱码文件和目录的实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux下删除乱码文件和目录方法1方法2总结Linux下删除乱码文件和目录方法1使用ls -i命令找到文件或目录

SpringBoot+EasyExcel实现自定义复杂样式导入导出

《SpringBoot+EasyExcel实现自定义复杂样式导入导出》这篇文章主要为大家详细介绍了SpringBoot如何结果EasyExcel实现自定义复杂样式导入导出功能,文中的示例代码讲解详细,... 目录安装处理自定义导出复杂场景1、列不固定,动态列2、动态下拉3、自定义锁定行/列,添加密码4、合并

mybatis执行insert返回id实现详解

《mybatis执行insert返回id实现详解》MyBatis插入操作默认返回受影响行数,需通过useGeneratedKeys+keyProperty或selectKey获取主键ID,确保主键为自... 目录 两种方式获取自增 ID:1. ​​useGeneratedKeys+keyProperty(推

Spring Boot集成Druid实现数据源管理与监控的详细步骤

《SpringBoot集成Druid实现数据源管理与监控的详细步骤》本文介绍如何在SpringBoot项目中集成Druid数据库连接池,包括环境搭建、Maven依赖配置、SpringBoot配置文件... 目录1. 引言1.1 环境准备1.2 Druid介绍2. 配置Druid连接池3. 查看Druid监控

Linux在线解压jar包的实现方式

《Linux在线解压jar包的实现方式》:本文主要介绍Linux在线解压jar包的实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux在线解压jar包解压 jar包的步骤总结Linux在线解压jar包在 Centos 中解压 jar 包可以使用 u

c++ 类成员变量默认初始值的实现

《c++类成员变量默认初始值的实现》本文主要介绍了c++类成员变量默认初始值,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录C++类成员变量初始化c++类的变量的初始化在C++中,如果使用类成员变量时未给定其初始值,那么它将被

Qt使用QSqlDatabase连接MySQL实现增删改查功能

《Qt使用QSqlDatabase连接MySQL实现增删改查功能》这篇文章主要为大家详细介绍了Qt如何使用QSqlDatabase连接MySQL实现增删改查功能,文中的示例代码讲解详细,感兴趣的小伙伴... 目录一、创建数据表二、连接mysql数据库三、封装成一个完整的轻量级 ORM 风格类3.1 表结构

基于Python实现一个图片拆分工具

《基于Python实现一个图片拆分工具》这篇文章主要为大家详细介绍了如何基于Python实现一个图片拆分工具,可以根据需要的行数和列数进行拆分,感兴趣的小伙伴可以跟随小编一起学习一下... 简单介绍先自己选择输入的图片,默认是输出到项目文件夹中,可以自己选择其他的文件夹,选择需要拆分的行数和列数,可以通过

Python中将嵌套列表扁平化的多种实现方法

《Python中将嵌套列表扁平化的多种实现方法》在Python编程中,我们常常会遇到需要将嵌套列表(即列表中包含列表)转换为一个一维的扁平列表的需求,本文将给大家介绍了多种实现这一目标的方法,需要的朋... 目录python中将嵌套列表扁平化的方法技术背景实现步骤1. 使用嵌套列表推导式2. 使用itert

Python使用pip工具实现包自动更新的多种方法

《Python使用pip工具实现包自动更新的多种方法》本文深入探讨了使用Python的pip工具实现包自动更新的各种方法和技术,我们将从基础概念开始,逐步介绍手动更新方法、自动化脚本编写、结合CI/C... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核