五、SRS5.0 分析（2）SrsServerAdapter分析

2024-01-28 21:36
文章标签 分析 srs5.0 srsserveradapter

本文主要是介绍五、SRS5.0 分析（2）SrsServerAdapter分析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

SrsServerAdapter最终的行为是建立 RTMP 连接，并完成推拉流的处理。

1、ISrsHybridServer接口

在上篇文章中分析到，一个服务启动的过程：
（1）创建注册RTMP、SRT、WebRTC服务
（2）初始化注册的服务
（3）启动注册进来的服务（调它们的 run 接口方法）

这是接口定义的规范：

class ISrsHybridServer
{
public:ISrsHybridServer(); //创建virtual ~ISrsHybridServer();
public:// Only ST initialized before each server, we could fork processes as such.virtual srs_error_t initialize() = 0; //初始化// Run each server, should never block except the SRS master server.virtual srs_error_t run(SrsWaitGroup* wg) = 0; // 启动// Stop each server, should do cleanup, for example, kill processes forked by server.virtual void stop() = 0;
};

class SrsServerAdapter : public ISrsHybridServer
{
private:SrsServer* srs;
public:SrsServerAdapter();virtual ~SrsServerAdapter();
public:virtual srs_error_t initialize();virtual srs_error_t run(SrsWaitGroup* wg);virtual void stop();
public:virtual SrsServer* instance();
};

2、SrsServerAdapter

  1. 构造函数、析构函数、初始化
SrsServerAdapter::SrsServerAdapter()
{srs = new SrsServer();
}SrsServerAdapter::~SrsServerAdapter()
{srs_freep(srs);
}srs_error_t SrsServerAdapter::initialize()
{srs_error_t err = srs_success;return err;
}

SrsServer是 RTMP 服务主模块。

  1. SrsServerAdapter::run方法
srs_error_t SrsServerAdapter::run(SrsWaitGroup* wg)
{srs_error_t err = srs_success;// Initialize the whole system, set hooks to handle server level events.if ((err = srs->initialize()) != srs_success) {return srs_error_wrap(err, "server initialize");}if ((err = srs->initialize_st()) != srs_success) {return srs_error_wrap(err, "initialize st");}if ((err = srs->initialize_signal()) != srs_success) {return srs_error_wrap(err, "initialize signal");}//⚠️监听if ((err = srs->listen()) != srs_success) {return srs_error_wrap(err, "listen");}if ((err = srs->register_signal()) != srs_success) {return srs_error_wrap(err, "register signal");}if ((err = srs->http_handle()) != srs_success) {return srs_error_wrap(err, "http handle");}if ((err = srs->ingest()) != srs_success) {return srs_error_wrap(err, "ingest");}if ((err = srs->start(wg)) != srs_success) {return srs_error_wrap(err, "start");}#ifdef SRS_GB28181if ((err = _srs_gb_manager->start()) != srs_success) {return srs_error_wrap(err, "start manager");}
#endifSrsLazySweepGc* gc = dynamic_cast<SrsLazySweepGc*>(_srs_gc);if ((err = gc->start()) != srs_success) {return srs_error_wrap(err, "start gc");}return err;
}

可以看到具体的逻辑委托给SrsServer，下面直接分析 SrsServer

3、SrsServer

盗一张图
在这里插入图片描述
左上角就是 SrsServer 部分。

  1. SrsServer::listen()

上面的 SrsServerAdapter::run方法中会调用到SrsServer::listen()

srs_error_t SrsServer::listen(){//1、创建 RTMP Listener// _srs_config->get_listens()获取RTMP.conf中的listen字段：1935，也就是端口号// SrsMultipleTcpListeners::add方法会创建SrsTcpListener，然后把它添加到内部维护的// 一个监听器列表里面：// 		std::vector<SrsTcpListener*> listeners_; // SrsTcpListener 后面会分析，此处暂且理解为进行 TCP 连接的一个基础类，TCP连接监听器。rtmp_listener_->add(_srs_config->get_listens())->set_label("RTMP");// 遍历列表listeners_，调用监听器的listen方法，进行监听。// 此处是进行 TCP 连接的监听，因为 RTMP 连接需要先建立好 TCP连接。if ((err = rtmp_listener_->listen()) != srs_success) {return srs_error_wrap(err, "rtmp listen");}...}
  • TCP 连接监听（RTMP连接需要先建立好 TCP 连接）

SrsTcpListener::listen()

srs_error_t SrsTcpListener::listen() {// srs_tcp_listen大致流程：// 1、创建socket// 2、bind 绑定服务器地址结构// 3、listen 设置监听上限if ((err = srs_tcp_listen(ip, port_, &lfd)) != srs_success) {return srs_error_wrap(err, "listen at %s:%d", ip.c_str(), port_);}// 2、创建协程，很显然 accpet 的逻辑，使用协程进行处理。srs_freep(trd);trd = new SrsSTCoroutine("tcp", this);if ((err = trd->start()) != srs_success) {return srs_error_wrap(err, "start coroutine");}	
}

 ⚠️ 注意srs_tcp_listen里面没有调用 accept，它的处理比较特殊，一般来说：1、如果server socket设置为阻塞，那么调用 accept 就会使线程阻塞，阻塞监听客户端的请求。2、如果server socket设置为非阻塞，那么需要不断循环检测客户端的请求，某种程度上也造成线程阻塞。一般都是用 IO 多路复用模型来处理这一块的逻辑，使用 epoll 模型监听事件。所以说，对于 accept 的处理的方式，影响着服务器的性能，SRS 使用协程来处理此处的逻辑。

SrsSTCoroutine是封装的协程类，但具体的逻辑委托给了SrsFastCoroutine

class SrsFastCoroutine
{
...
public:srs_error_t start();void stop();void interrupt();inline srs_error_t pull() {if (trd_err == srs_success) {return srs_success;}return srs_error_copy(trd_err);}const SrsContextId& cid();virtual void set_cid(const SrsContextId& cid);
...
}

上面罗列了SrsFastCoroutine对外提供的方法，通过分析对外提供的方法，知道其用途，这样来读代码，省时间且有针对性，不可能每个地方都花时间搞懂，用不上就没必要，因为读懂 SRS 也是为二次开发。

比如上面代码：trd->start()，就是启动协程，通过追踪发现，最终还是调用的st_thread_create，该函数是StateThreads库提供的方法，具体的StateThreads用法会专门在一篇文章讲解。暂时理解为 pthread_create()。

下面是StateThreads库相关资料：

API 文档：https://state-threads.sourceforge.net/docs/reference.html
https://blog.csdn.net/win_lin/article/details/8242653
https://www.xianwaizhiyin.net/?cat=24
https://srs.xianwaizhiyin.net/state-threads/debug.html

trd->start()启动协程会执行到SrsFastCoroutine::cycle()，因为trd构造的时候传入
SrsTcpListener的指针，所以执行到SrsTcpListener::cycle()方法：处理 accept 的逻辑, 开始监听客户端连接。

srs_error_t SrsTcpListener::cycle()
{srs_error_t err = srs_success;while (true) {if ((err = trd->pull()) != srs_success) {return srs_error_wrap(err, "tcp listener");}srs_netfd_t fd = srs_accept(lfd, NULL, NULL, SRS_UTIME_NO_TIMEOUT);if(fd == NULL){return srs_error_new(ERROR_SOCKET_ACCEPT, "accept at fd=%d", srs_netfd_fileno(lfd));}if ((err = srs_fd_closeexec(srs_netfd_fileno(fd))) != srs_success) {return srs_error_wrap(err, "set closeexec");}if ((err = handler->on_tcp_client(this, fd)) != srs_success) {return srs_error_wrap(err, "handle fd=%d", srs_netfd_fileno(fd));}}return err;
}

当执行到srs_accept时，该线程就阻塞监听了，直到客户端连接时唤醒。

至此，TCP 的监听逻辑分析结束。

每一个监听对应一个协程
每一个连接也对应一个协程
类似的，之前我们在网络编程中，每一个连接对应一个线程。
  • 处理客户端的连接
    在这里插入图片描述
    使用 RTMP 客户端进行连接，这边监听到连接请求，解除阻塞。
    会执行到 handler->on_tcp_client，然后最终执行到：SrsServer::on_tcp_client方法
srs_error_t SrsServer::do_on_tcp_client(ISrsListener* listener, srs_netfd_t& stfd)
{	...if (!resource) {if (listener == rtmp_listener_) {// SrsRtmpConn 是用于建立 RTMP 连接的类resource = new SrsRtmpConn(this, stfd2, ip, port);...}//将连接对象conn加入SrsResourceManager统一管理conn_manager->add(resource);// 启动连接对象conn内部的协程// SrsRtmpConn对象启动的协程，执行的方法是SrsRtmpConn::do_cycle()ISrsStartable* conn = dynamic_cast<ISrsStartable*>(resource);if ((err = conn->start()) != srs_success) {return srs_error_wrap(err, "start conn coroutine");}return err;  
}

SrsRtmpConn::do_cycle()

主要完成RTMP协议的（1）握手过程；（2）建立连接；（3）请求推(拉)流，具体处理如下：

srs_error_t SrsRtmpConn::do_cycle()
{	...// （1）RTMP 握手if ((err = rtmp->handshake()) != srs_success) {return srs_error_wrap(err, "rtmp handshake");}...//（2）建立连接，RTMP ConnectSrsRequest* req = info->req;if ((err = rtmp->connect_app(req)) != srs_success) {return srs_error_wrap(err, "rtmp connect tcUrl");}...//（3）请求推拉流if ((err = service_cycle()) != srs_success) {err = srs_error_wrap(err, "service cycle");}...
}

这里涉及到 RTMP 建立连接的过程，不细说，可以参考：三、RTMP推流的原理及实现

RTMP 的handshake和connect_app不细说了，下面service_cycle讲一下，因为我也不熟。

参考：

https://blog.csdn.net/adkada1/article/details/120590511

SrsRtmpConn::service_cycle()

srs_error_t SrsRtmpConn::service_cycle(){// （1）获取服务端默认配置的 ack size = 2500000 字节int out_ack_size = _srs_config->get_out_ack_size(req->vhost);// （2）向客户端发送Window Acknowledgement Size消息（接收 buffer）//  并设置本端的in_window_ack_size参数（发送 buffer）if (out_ack_size && (err = rtmp->set_window_ack_size(out_ack_size)) != srs_success) {return srs_error_wrap(err, "rtmp: set out window ack size");}// （3）向客户端发送Set Peer Bandwidth消息。if ((err = rtmp->set_peer_bandwidth((int)(2.5 * 1000 * 1000), 2)) != srs_success) {return srs_error_wrap(err, "rtmp: set peer bandwidth");}// (4) 向客户端发送 Set Chunk Size消息，这里设置的大小 60000 字节，工程经验。int chunk_size = _srs_config->get_chunk_size(req->vhost);if ((err = rtmp->set_chunk_size(chunk_size)) != srs_success) {return srs_error_wrap(err, "rtmp: set chunk size %d", chunk_size);}// （5）回复客户端 connect ok// response the client connect ok.if ((err = rtmp->response_connect_app(req, local_ip.c_str())) != srs_success) {return srs_error_wrap(err, "rtmp: response connect app");}
}

  • RTMP 的Window Acknowledgement Size (ID=5) 消息
    Window Acknowledgement Size是⽤于设置窗⼝确认⼤⼩的消息。
    Window Acknowledgement Size消息是在双端会话开始时发送，一般是建立起链接之后。用于指明期望获得确认的大小。

  • RTMP 的Set Peer Bandwidth (ID=6) 消息
    设置对端带宽的消息。
    客户端或服务端发送本消息更新对等端的输出带宽。
    接收端在收到发送端这个消息后，会通过设置消息中的Window Acknowledgement Size来限制发送端的发送带宽。
    如果消息中的Window Acknowledgement Size与上⼀次发送给发送端的size不同的话，接收端要回馈⼀个Window
    Acknowledgement Size的控制消息给发送端。

  • RTMP 的Set Chunk Size(ID=1) 消息
    用于设置 chunk 中 Data 字段的最大字节数，默认为 128 字节，通信过程中可以通过发送该消息来设置 chunk size。
    通信双方会各自独立维护一个 chunk size。比如：当 A 向 B 发送一个 200 字节的 Message ，但是 A默认的 chunk size是 128 字节，那就要把 这个 Message 拆成 Data 为 128字节和 72 字节的两个 chunk发送。

上面完成了 RTMP 连接的过程，至此 RTMP 连接成功。

SrsRtmpConn::service_cycle()

stream_service_cycle 根据接收到的报文，区分推流端和拉流端，完成推、拉流的创建过程

srs_error_t SrsRtmpConn::service_cycle()
{while (true) {if ((err = trd->pull()) != srs_success) {return srs_error_wrap(err, "rtmp: thread quit");}err = stream_service_cycle();...}
}

stream_service_cycle

srs_error_t SrsRtmpConn::stream_service_cycle()
{//...//对于推流端，每个推流端通过fetch_or_create函数生成一个对应的SrsLiveSource对象//SrsLiveSource暂时还没有了解是干嘛用的SrsLiveSource* source = NULL;if ((err = _srs_sources->fetch_or_create(req, server, &source)) != srs_success) {return srs_error_wrap(err, "rtmp: fetch source");}// info->type是SrsRtmpConnFMLEPublish，很显然，这推流操作switch (info->type) {switch (info->type) {case SrsRtmpConnPlay:  // 拉流客户端类型...return err;case SrsRtmpConnFMLEPublish:  // 标准FFMPEG推流端类型if ((err = rtmp->start_fmle_publish(info->res->stream_id)) != srs_success) {return srs_error_wrap(err, "rtmp: start FMLE publish");}case SrsRtmpConnHaivisionPublish:   // 海康设备推流...case SrsRtmpConnFlashPublish:  // flash推流...}
}

我调试的工程属于是 FFMPEG 推流，属于SrsRtmpConnFMLEPublish，所以执行到 rtmp->start_fmle_publish，该方法会对客户端发送过来的消息，比如 FCPublish、createStream、publish的消息做出应答（下图中的 onFCPublish、_result 、onStatus），表示同意建立起 RTMP 连接，并且可以开始推流。

接下来在 SrsRtmpConn::publishing方法中会处理推流的逻辑，下一章再具体分析。

在这里插入图片描述

4、总结

本文主要讲解SrsServerAdapter，最终的行为是建立 RTMP 连接，并完成推拉流的处理。
本文忽略了很多里面的逻辑，这些逻辑是无数次迭代出来的，属于工程经验的总结，所以看不是很懂，当然也不必纠结。我认为先把握主要的逻辑为主。

这篇关于五、SRS5.0 分析（2）SrsServerAdapter分析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/654836。

相关文章

性能分析之MySQL索引实战案例

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题，本文将一起从代码层去分析为什么没有建立索引？ 开源ERP项目地址：https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置
阅读更多...
SWAP作物生长模型安装教程、数据制备、敏感性分析、气候变化影响、R模型敏感性分析与贝叶斯优化、Fortran源代码分析、气候数据降尺度与变化影响分析

SWAP作物生长模型安装教程、数据制备、敏感性分析、气候变化影响、R模型敏感性分析与贝叶斯优化、Fortran源代码分析、气候数据降尺度与变化影响分析

查看原文>>>全流程SWAP农业模型数据制备、敏感性分析及气候变化影响实践技术应用 SWAP模型是由荷兰瓦赫宁根大学开发的先进农作物模型，它综合考虑了土壤-水分-大气以及植被间的相互作用；是一种描述作物生长过程的一种机理性作物生长模型。它不但运用Richard方程，使其能够精确的模拟土壤中水分的运动，而且耦合了WOFOST作物模型使作物的生长描述更为科学。 本文让更多的科研人员和农业工作者
阅读更多...
MOLE 2.5 分析分子通道和孔隙

MOLE 2.5 分析分子通道和孔隙

软件介绍 生物大分子通道和孔隙在生物学中发挥着重要作用，例如在分子识别和酶底物特异性方面。 我们介绍了一种名为 MOLE 2.5 的高级软件工具，该工具旨在分析分子通道和孔隙。 与其他可用软件工具的基准测试表明，MOLE 2.5 相比更快、更强大、功能更丰富。作为一项新功能，MOLE 2.5 可以估算已识别通道的物理化学性质。 软件下载 https://pan.quark.cn/s/57
阅读更多...
衡石分析平台使用手册-单机安装及启动

衡石分析平台使用手册-单机安装及启动

单机安装及启动​ 本文讲述如何在单机环境下进行 HENGSHI SENSE 安装的操作过程。 在安装前请确认网络环境，如果是隔离环境，无法连接互联网时，请先按照 离线环境安装依赖的指导进行依赖包的安装，然后按照本文的指导继续操作。如果网络环境可以连接互联网，请直接按照本文的指导进行安装。 准备工作​ 请参考安装环境文档准备安装环境。 配置用户与安装目录。 在操作前请检查您是否有 sud
阅读更多...
线性因子模型 - 独立分量分析（ICA）篇

线性因子模型 - 独立分量分析（ICA）篇

序言 线性因子模型是数据分析与机器学习中的一类重要模型，它们通过引入潜变量（ latent variables \text{latent variables} latent variables）来更好地表征数据。其中，独立分量分析（ ICA \text{ICA} ICA）作为线性因子模型的一种，以其独特的视角和广泛的应用领域而备受关注。 ICA \text{ICA} ICA旨在将观察到的复杂信号
阅读更多...
【软考】希尔排序算法分析

【软考】希尔排序算法分析

目录 1. c代码2. 运行截图3. 运行解析 1. c代码 #include <stdio.h>#include <stdlib.h> void shellSort(int data[], int n){// 划分的数组，例如8个数则为[4, 2, 1]int *delta;int k;// i控制delta的轮次int i;// 临时变量，换值int temp;in
阅读更多...
三相直流无刷电机(BLDC)控制算法实现：BLDC有感启动算法思路分析

三相直流无刷电机(BLDC)控制算法实现：BLDC有感启动算法思路分析

一枚从事路径规划算法、运动控制算法、BLDC/FOC电机控制算法、工控、物联网工程师，爱吃土豆。如有需要技术交流或者需要方案帮助、需求：以下为联系方式—V 方案1：通过霍尔传感器IO中断触发换相 1.1 整体执行思路 霍尔传感器U、V、W三相通过IO+EXIT中断的方式进行霍尔传感器数据的读取。将IO口配置为上升沿+下降沿中断触发的方式。当霍尔传感器信号发生发生信号的变化就会触发中断在中断
阅读更多...
kubelet组件的启动流程源码分析

kubelet组件的启动流程源码分析

概述 摘要: 本文将总结kubelet的作用以及原理，在有一定基础认识的前提下，通过阅读kubelet源码，对kubelet组件的启动流程进行分析。 正文 kubelet的作用 这里对kubelet的作用做一个简单总结。 节点管理 节点的注册 节点状态更新 容器管理(pod生命周期管理) 监听apiserver的容器事件 容器的创建、删除(CRI) 容器的网络的创建与删除
阅读更多...
PostgreSQL核心功能特性与使用领域及场景分析

PostgreSQL核心功能特性与使用领域及场景分析

PostgreSQL有什么优点？ 开源和免费 PostgreSQL是一个开源的数据库管理系统，可以免费使用和修改。这降低了企业的成本，并为开发者提供了一个活跃的社区和丰富的资源。 高度兼容 PostgreSQL支持多种操作系统（如Linux、Windows、macOS等）和编程语言（如C、C++、Java、Python、Ruby等），并提供了多种接口（如JDBC、ODBC、ADO.NET等
阅读更多...
OpenCV结构分析与形状描述符（11）椭圆拟合函数fitEllipse()的使用

OpenCV结构分析与形状描述符（11）椭圆拟合函数fitEllipse()的使用

操作系统：ubuntu22.04 OpenCV版本：OpenCV4.9 IDE:Visual Studio Code 编程语言：C++11 算法描述 围绕一组2D点拟合一个椭圆。 该函数计算出一个椭圆，该椭圆在最小二乘意义上最好地拟合一组2D点。它返回一个内切椭圆的旋转矩形。使用了由[90]描述的第一个算法。开发者应该注意，由于数据点靠近包含的 Mat 元素的边界，返回的椭圆/旋转矩形数据
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2