1.目标

学习 RTMP 端口监听模块前，需明确两个核心问题：

  1. 如何从配置文件读取对应的端口？：服务器如何解析配置文件，获取正确服务端口。
  2. 如何启动 RTMP 监听并处理业务逻辑？：服务器如何绑定端口，等待客户端连接。创建rtmpconn连接对象，包含有握手和断开等等

1.1 引入服务启动时序图

通过此引入可以知道先进行配置文件解析，其次开启 SrsServer服务！

从main程序开始到SrsServer服务启动的时序图如下：

        

2.端口获取

2.1 核心调用栈

下列分别是rtmp端口获取操作调用栈。分别是检查配置文件是否正常，检查最大连接数量。

(gdb) bt
#0  SrsConfig::get_listens[abi:cxx11]() (this=0x555555e90340) at src/app/srs_app_config.cpp:3135
#1  0x0000555555737fbe in SrsConfig::check_normal_config (this=0x555555e90340) at src/app/srs_app_config.cpp:2559
#2  0x0000555555736af3 in SrsConfig::check_config (this=0x555555e90340) at src/app/srs_app_config.cpp:2431
#3  0x0000555555845bda in do_main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:184
#4  0x0000555555845eb2 in main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:216


(gdb) bt
#0  SrsConfig::get_listens[abi:cxx11]() (this=0x555555e90340) at src/app/srs_app_config.cpp:3136
#1  0x000055555573c7b3 in SrsConfig::check_number_connections (this=0x555555e90340) at src/app/srs_app_config.cpp:2916
#2  0x0000555555736b4c in SrsConfig::check_config (this=0x555555e90340) at src/app/srs_app_config.cpp:2435
#3  0x0000555555845bda in do_main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:184
#4  0x0000555555845eb2 in main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:216

2.2 核心API 

该api是从文件解析出listen port 存储起来。后续会对每个端口new一个SrsBufferListener。

vector<string> SrsConfig::get_listens()
{
    std::vector<string> ports;
    
    SrsConfDirective* conf = root->get("listen");
    if (!conf) {
        return ports;
    }
    
    for (int i = 0; i < (int)conf->args.size(); i++) {
        ports.push_back(conf->args.at(i));
    }
    
    return ports;
}

3.端口监听

       流程顺序：端口创建SrsBufferListener再创建SrsTcpListener，SrsTcpListener引出协程对象并开启协程循环。通过这种方法实现了一连接一协程收发数据！

  1. SrsServer每个端口  1:1  SrsBufferListener。这一层是协议的解耦，基类对象public SrsListener
  2. SrsBufferListener  1:1  SrsTcpListener。这一层是tcp协议，并包含协程设计。
  3. SrsTcpListener 1:1 SrsSTCoroutine。这一层是开启协程开启 trd->start() ->>实际调用

    SrsTcpListener::cycle()

3.1 调用栈

(gdb) bt
#0  SrsServer::listen_rtmp (this=0x555555f498f0) at src/app/srs_app_server.cpp:1177
#1  0x00005555556d37b7 in SrsServer::listen (this=0x555555ed3740) at src/app/srs_app_server.cpp:757
#2  0x00005555557d750b in SrsServerAdapter::run (this=0x555555ed59e0, wg=0x7fffffffdc80) at src/app/srs_app_hybrid.cpp:160
#3  0x00005555557d8072 in SrsHybridServer::run (this=0x555555ebe9d0) at src/app/srs_app_hybrid.cpp:275
#4  0x0000555555847917 in run_hybrid_server () at src/main/srs_main_server.cpp:497
#5  0x00005555558473da in run_directly_or_daemon () at src/main/srs_main_server.cpp:423
#6  0x0000555555845c2d in do_main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:201
#7  0x0000555555845eb2 in main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:216

(gdb) bt
#0  SrsBufferListener::listen (this=0x55555572259e <SrsFileLog::trace(char const*, _SrsContextId, char const*, ...)>, i="", p=21845) at src/app/srs_app_server.cpp:89
#1  0x00005555556d6cf7 in SrsServer::listen_rtmp (this=0x555555ed3740) at src/app/srs_app_server.cpp:1193
#2  0x00005555556d37b7 in SrsServer::listen (this=0x555555ed3740) at src/app/srs_app_server.cpp:757
#3  0x00005555557d750b in SrsServerAdapter::run (this=0x555555ed59e0, wg=0x7fffffffdc80) at src/app/srs_app_hybrid.cpp:160
#4  0x00005555557d8072 in SrsHybridServer::run (this=0x555555ebe9d0) at src/app/srs_app_hybrid.cpp:275
#5  0x0000555555847917 in run_hybrid_server () at src/main/srs_main_server.cpp:497
#6  0x00005555558473da in run_directly_or_daemon () at src/main/srs_main_server.cpp:423
#7  0x0000555555845c2d in do_main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:201
#8  0x0000555555845eb2 in main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:216


(gdb) bt
#0  SrsTcpListener::listen (this=0x7fffffffda10) at src/app/srs_app_listener.cpp:237
#1  0x00005555556cf354 in SrsBufferListener::listen (this=0x555555f20500, i="0.0.0.0", p=1935) at src/app/srs_app_server.cpp:98
#2  0x00005555556d6cf7 in SrsServer::listen_rtmp (this=0x555555ed3740) at src/app/srs_app_server.cpp:1193
#3  0x00005555556d37b7 in SrsServer::listen (this=0x555555ed3740) at src/app/srs_app_server.cpp:757
#4  0x00005555557d750b in SrsServerAdapter::run (this=0x555555ed59e0, wg=0x7fffffffdc80) at src/app/srs_app_hybrid.cpp:160
#5  0x00005555557d8072 in SrsHybridServer::run (this=0x555555ebe9d0) at src/app/srs_app_hybrid.cpp:275
#6  0x0000555555847917 in run_hybrid_server () at src/main/srs_main_server.cpp:497
#7  0x00005555558473da in run_directly_or_daemon () at src/main/srs_main_server.cpp:423
#8  0x0000555555845c2d in do_main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:201
#9  0x0000555555845eb2 in main (argc=3, argv=0x7fffffffe1c8) at src/main/srs_main_server.cpp:216

3.2 核心API 

从上到下调用

srs_error_t SrsServer::listen_rtmp()
{
    srs_error_t err = srs_success;
    
    // stream service port.
    std::vector<std::string> ip_ports = _srs_config->get_listens();
    srs_assert((int)ip_ports.size() > 0);
    
    close_listeners(SrsListenerRtmpStream);
    
    for (int i = 0; i < (int)ip_ports.size(); i++) {
        SrsListener* listener = new SrsBufferListener(this, SrsListenerRtmpStream);//每个端口对应一个SrsBufferListener
        listeners.push_back(listener);

        int port; string ip;
        srs_parse_endpoint(ip_ports[i], ip, port);
        
        if ((err = listener->listen(ip, port)) != srs_success) {
            srs_error_wrap(err, "rtmp listen %s:%d", ip.c_str(), port);
        }
    }
    
    return err;
}

srs_error_t SrsBufferListener::listen(string i, int p)
{
    srs_error_t err = srs_success;
    
    ip = i;
    port = p;
    
    srs_freep(listener);
    listener = new SrsTcpListener(this, ip, port);//每个SrsBufferListener对应一个SrsTcpListener
    
    if ((err = listener->listen()) != srs_success) {
        return srs_error_wrap(err, "buffered tcp listen");
    }
    
    string v = srs_listener_type2string(type);
    srs_trace("%s listen at tcp://%s:%d, fd=%d", v.c_str(), ip.c_str(), port, listener->fd());
    
    return err;
}


srs_error_t SrsTcpListener::listen()
{
    srs_error_t err = srs_success;

    if ((err = srs_tcp_listen(ip, port, &lfd)) != srs_success) {
        return srs_error_wrap(err, "listen at %s:%d", ip.c_str(), port);
    }
    
    srs_freep(trd);
    trd = new SrsSTCoroutine("tcp", this); //每个SrsTcpListener对应SrsSTCoroutine负责accept
    if ((err = trd->start()) != srs_success) {
        return srs_error_wrap(err, "start coroutine");
    }
    
    return err;
}

#最底层调用
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);//等待新的fd
        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) {//设置fd为close-on-exec，防止子进程继承该文件描述符
            return srs_error_wrap(err, "set closeexec");
        }
        
        if ((err = handler->on_tcp_client(fd)) != srs_success) {//handler是上层对象，SrsBufferListener::on_tcp_client or SrsHttpFlvListener::on_tcp_client
            return srs_error_wrap(err, "handle fd=%d", srs_netfd_fileno(fd));
        }
    }
    
    return err;
}


#开始往上层回调
srs_error_t SrsBufferListener::on_tcp_client(srs_netfd_t stfd)
{
    srs_error_t err = server->accept_client(type, stfd);//最终RTMP业务回调到SrsServer::accept_client
    if (err != srs_success) {
        srs_warn("accept client failed, err is %s", srs_error_desc(err).c_str());
        srs_freep(err);
    }
    
    return srs_success;
}

4. 新连接 

根据上文，底层回调上来的fd交到应用层来处理。这里以rtmp举例，主要是

启动 RTMP 监听器→接收新 TCP 连接→创建SrsRtmpConn连接对象→协程驱动cycle()主循环→完成握手、应用连接、媒体流传输→连接断开清理。

#主要是根据type创建对应连接
srs_error_t SrsServer::accept_client(SrsListenerType type, srs_netfd_t stfd)
{
    srs_error_t err = srs_success;
    
    ISrsStartableConneciton* conn = NULL;
    //fd_to_resource：会带出一个连接对象 conn
    if ((err = fd_to_resource(type, stfd, &conn)) != srs_success) {
        if (srs_error_code(err) == ERROR_SOCKET_GET_PEER_IP && _srs_config->empty_ip_ok())             
    {
            srs_close_stfd(stfd); srs_error_reset(err);
            return srs_success;
        }
        return srs_error_wrap(err, "fd to resource");
    }
    srs_assert(conn);
    
    // directly enqueue, the cycle thread will remove the client.
    conn_manager->add(conn);
    //根据连接对象开启协程
    if ((err = conn->start()) != srs_success) {
        return srs_error_wrap(err, "start conn coroutine");
    }
    
    return err;
}


srs_error_t SrsServer::fd_to_resource(SrsListenerType type, srs_netfd_t stfd, ISrsStartableConneciton** pr)
{
    srs_error_t err = srs_success;
    
    int fd = srs_netfd_fileno(stfd);
    string ip = srs_get_peer_ip(fd);
    int port = srs_get_peer_port(fd);
.........................................
.........................................
    if (type == SrsListenerRtmpStream) {
        *pr = new SrsRtmpConn(this, stfd, ip, port);
    } else if (type == SrsListenerHttpApi) {
        *pr = new SrsHttpApi(false, this, stfd, http_api_mux, ip, port);
    } else if (type == SrsListenerHttpsApi) {
        *pr = new SrsHttpApi(true, this, stfd, http_api_mux, ip, port);
    } else if (type == SrsListenerHttpStream) {
        *pr = new SrsResponseOnlyHttpConn(false, this, stfd, http_server, ip, port);
    } else if (type == SrsListenerHttpsStream) {
        *pr = new SrsResponseOnlyHttpConn(true, this, stfd, http_server, ip, port);
    } else {
        srs_warn("close for no service handler. fd=%d, ip=%s:%d", fd, ip.c_str(), port);
        srs_close_stfd(stfd);
        return err;
    }
    
    return err;
}

#rtmp逻辑处理
srs_error_t SrsRtmpConn::cycle()
{
    srs_error_t err = do_cycle();

    // Notify manager to remove it.
    // Note that we create this object, so we use manager to remove it.
    manager->remove(this);

    // success.
    if (err == srs_success) {
        srs_trace("client finished.");
        return err;
    }

    // It maybe success with message.
    if (srs_error_code(err) == ERROR_SUCCESS) {
        srs_trace("client finished%s.", srs_error_summary(err).c_str());
        srs_freep(err);
        return err;
    }

    // client close peer.
    // TODO: FIXME: Only reset the error when client closed it.
    if (srs_is_client_gracefully_close(err)) {
        srs_warn("client disconnect peer. ret=%d", srs_error_code(err));
    } else if (srs_is_server_gracefully_close(err)) {
        srs_warn("server disconnect. ret=%d", srs_error_code(err));
    } else {
        srs_error("serve error %s", srs_error_desc(err).c_str());
    }

    srs_freep(err);
    return srs_success;
}


// 负责处理一个RTMP客户端连接的生命周期，包括连接建立、握手、应用连接、服务循环以及断开连接等步骤。
srs_error_t SrsRtmpConn::do_cycle()
{
    srs_error_t err = srs_success;
    
    // 打印RTMP客户端的IP地址和端口
    srs_trace("RTMP client ip=%s:%d, fd=%d", ip.c_str(), port, srs_netfd_fileno(stfd));
    
    // 设置RTMP的接收和发送超时时间
    rtmp->set_recv_timeout(SRS_CONSTS_RTMP_TIMEOUT);
    rtmp->set_send_timeout(SRS_CONSTS_RTMP_TIMEOUT);

    // 执行RTMP握手
    if ((err = rtmp->handshake()) != srs_success) {
        return srs_error_wrap(err, "rtmp handshake");
    }

    // 获取RTMP代理的真实客户端IP地址
    uint32_t rip = rtmp->proxy_real_ip();
    if (rip > 0) {
        srs_trace("RTMP proxy real client ip=%d.%d.%d.%d",
            uint8_t(rip>>24), uint8_t(rip>>16), uint8_t(rip>>8), uint8_t(rip));
    }
    
    // 获取请求信息
    SrsRequest* req = info->req;
    if ((err = rtmp->connect_app(req)) != srs_success) {
        return srs_error_wrap(err, "rtmp connect tcUrl");
    }
    
    // 将客户端IP地址设置到请求中
    // set client ip to request.
    req->ip = ip;
    
    // 打印连接应用的信息
    srs_trace("connect app, tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%d, app=%s, args=%s",
        req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(),
        req->schema.c_str(), req->vhost.c_str(), req->port,
        req->app.c_str(), (req->args? "(obj)":"null"));
    
    // 显示客户端身份信息
    // show client identity
    if(req->args) {
        std::string srs_version;
        std::string srs_server_ip;
        int srs_pid = 0;
        int srs_id = 0;
        
        SrsAmf0Any* prop = NULL;
        if ((prop = req->args->ensure_property_string("srs_version")) != NULL) {
            srs_version = prop->to_str();
        }
        if ((prop = req->args->ensure_property_string("srs_server_ip")) != NULL) {
            srs_server_ip = prop->to_str();
        }
        if ((prop = req->args->ensure_property_number("srs_pid")) != NULL) {
            srs_pid = (int)prop->to_number();
        }
        if ((prop = req->args->ensure_property_number("srs_id")) != NULL) {
            srs_id = (int)prop->to_number();
        }
        
        if (srs_pid > 0) {
            srs_trace("edge-srs ip=%s, version=%s, pid=%d, id=%d",
                srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id);
        }
    }
    
    // 执行服务循环
    if ((err = service_cycle()) != srs_success) {
        err = srs_error_wrap(err, "service cycle");
    }
    
    srs_error_t r0 = srs_success;
    if ((r0 = on_disconnect()) != srs_success) {
        err = srs_error_wrap(err, "on disconnect %s", srs_error_desc(r0).c_str());
        srs_freep(r0);
    }
    
    // 如果客户端被重定向到其他服务器，则已经记录了该事件
    // If client is redirect to other servers, we already logged the event.
    if (srs_error_code(err) == ERROR_CONTROL_REDIRECT) {
        srs_error_reset(err);
    }
    
    return err;
}

5.总结

RTMP 监听的核心流程是：
SrsServer → 初始化 SrsBufferListener → 每个 SrsBufferListener 管理一个 SrsTcpListener → SrsTcpListener 通过协程循环接受新连接 → 连接通过 on_tcp_client 回调传递给 SrsServer 处理业务逻辑。on_tcp_client启动 RTMP 监听器→接收新 TCP 连接→创建SrsRtmpConn连接对象→协程驱动cycle()主循环→完成握手、应用连接、媒体流传输→连接断开清理。

下列给出程序开始到rtmp监听整体时序图：

学习资料分享

0voice · GitHub

# 服务器 # webrtc
Logo
乐奇 Rokid 开放社区

更多推荐

cover

让界面替我“摇“一次：一个AR祈愿智能体的AIUI实践

avatar 乐奇 Rokid 开放社区
cover

AR 智能眼镜并不是简单地在眼前放一块屏幕，而是把**摄像头、传感器、计算机视觉、空间定位、语音交互和云端服务**连接起来

avatar 乐奇 Rokid 开放社区

智能眼镜市场全景解析：从技术演进到产业格局

本文系统梳理智能眼镜市场，涵盖产品形态、技术演进、市场规模、产业链结构及竞争格局。指出智能眼镜正从音频交互向AR显示与空间计算演进，受AI大模型、供应链成熟与生态延伸驱动，呈现分层发展态势。虽面临场景刚需不足、续航重量矛盾与隐私挑战，但B端应用先行、AI原生设计与标准完善将推动行业迈向规模化渗透，未来需关注核心技术突破与真实使用数据。

avatar 乐奇 Rokid 开放社区