五、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。

相关文章

怎样通过分析GC日志来定位Java进程的内存问题

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》：本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.
阅读更多...
MySQL中的表连接原理分析

MySQL中的表连接原理分析

《MySQL中的表连接原理分析》：本文主要介绍MySQL中的表连接原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1、背景2、环境3、表连接原理【1】驱动表和被驱动表【2】内连接【3】外连接【4编程】嵌套循环连接【5】join buffer4、总结1、背景
阅读更多...
python中Hash使用场景分析

python中Hash使用场景分析

《python中Hash使用场景分析》Python的hash()函数用于获取对象哈希值,常用于字典和集合,不可变类型可哈希,可变类型不可,常见算法包括除法、乘法、平方取中和随机数哈希,各有优缺点,需根... 目录python中的 Hash除法哈希算法乘法哈希算法平方取中法随机数哈希算法小结在Python中，
阅读更多...
Java Stream的distinct去重原理分析

Java Stream的distinct去重原理分析

《JavaStream的distinct去重原理分析》Javastream中的distinct方法用于去除流中的重复元素,它返回一个包含过滤后唯一元素的新流,该方法会根据元素的hashcode和eq... 目录一、distinct 的基础用法与核心特性二、distinct 的底层实现原理1. 顺序流中的去重
阅读更多...
关于MyISAM和InnoDB对比分析

关于MyISAM和InnoDB对比分析

《关于MyISAM和InnoDB对比分析》：本文主要介绍关于MyISAM和InnoDB对比分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录开篇：从交通规则看存储引擎选择理解存储引擎的基本概念技术原理对比1. 事务支持：ACID的守护者2. 锁机制：并发控制的艺
阅读更多...
MyBatis Plus 中 update_time 字段自动填充失效的原因分析及解决方案(最新整理)

MyBatis Plus 中 update_time 字段自动填充失效的原因分析及解决方案(最新整理)

《MyBatisPlus中update_time字段自动填充失效的原因分析及解决方案(最新整理)》在使用MyBatisPlus时,通常我们会在数据库表中设置create_time和update... 目录前言一、问题现象二、原因分析三、总结：常见原因与解决方法对照表四、推荐写法前言在使用 MyBATis
阅读更多...
Python主动抛出异常的各种用法和场景分析

Python主动抛出异常的各种用法和场景分析

《Python主动抛出异常的各种用法和场景分析》在Python中,我们不仅可以捕获和处理异常,还可以主动抛出异常,也就是以类的方式自定义错误的类型和提示信息,这在编程中非常有用,下面我将详细解释主动抛... 目录一、为什么要主动抛出异常？二、基本语法：raise关键字基本示例三、raise的多种用法1. 抛
阅读更多...
github打不开的问题分析及解决

github打不开的问题分析及解决

《github打不开的问题分析及解决》：本文主要介绍github打不开的问题分析及解决,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、找到github.com域名解析的ip地址二、找到github.global.ssl.fastly.net网址解析的ip地址三
阅读更多...
Mysql的主从同步/复制的原理分析

Mysql的主从同步/复制的原理分析

《Mysql的主从同步/复制的原理分析》：本文主要介绍Mysql的主从同步/复制的原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录为什么要主从同步？mysql主从同步架构有哪些？Mysql主从复制的原理/整体流程级联复制架构为什么好？Mysql主从复制注意
阅读更多...
java -jar命令运行 jar包时运行外部依赖jar包的场景分析

java -jar命令运行 jar包时运行外部依赖jar包的场景分析

《java-jar命令运行jar包时运行外部依赖jar包的场景分析》：本文主要介绍java-jar命令运行jar包时运行外部依赖jar包的场景分析,本文给大家介绍的非常详细,对大家的学习或工作... 目录Java -jar命令运行 jar包时如何运行外部依赖jar包场景：解决：方法一、启动参数添加： -Xb
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2