[ Ngnix与高并发服务器 ] 【守望者 SRS 】SRS 源代码分析笔记（0.9.194）-分析服务器对端口的监听 ...

2014-10-12 16:16| 发布者: zhouy | 查看: 2067 | 收藏

摘要: 第一部分分析服务器对端口的监听端口监听与初始化（一）全局变量_srs_config 存放srs.conf中的配置指令。全局变量_srs_server SRS服务器的全局类对象。SrsServer中的listen_rtmpSrs_main_server.cpp run_master() ...

第一部分分析服务器对端口的监听端口监听与初始化（一）

全局变量_srs_config 存放srs.conf中的配置指令。

全局变量_srs_server SRS服务器的全局类对象。

SrsServer中的listen_rtmp

Srs_main_server.cpp run_master()函数中：

if ((ret = _srs_server->listen()) != ERROR_SUCCESS)

{

return ret;

}

看下SrsServer中的类listen函数是如何实现的：

int SrsServer::listen()

{

int ret = ERROR_SUCCESS;

if ((ret = listen_rtmp()) != ERROR_SUCCESS) {

return ret;

}

if ((ret = listen_http_api()) != ERROR_SUCCESS) {

return ret;

}

if ((ret = listen_http_stream()) != ERROR_SUCCESS) {

return ret;

}

return ret;

}

listen函数实现了对rtmp,http_api和http_stream的端口监听，现在进一步分析下listen_rtmp函数的实现过程，代码如下，其中红色笔者添加的注释，并且没跳注释就用数字标号，用于流程之间的传递：

int SrsServer::listen_rtmp()

{

int ret = ERROR_SUCCESS;

// stream service port.

//_srs_config全局变量，srs.conf中的所有配置指令都在对象root成员中。(1)

std::vector<std::string> ports = _srs_config->get_listen();//获取到端口

srs_assert((int)ports.size() > 0);

close_listeners(SrsListenerRtmpStream);

for (int i = 0; i < (int)ports.size(); i++) {

//对每个端口创建一个SrsListener实例，然后将实例指针放到SrsServer中的 std::vector<SrsListener*> listeners 中去;创建SrsListener实例的时候需要将SrsServer实例本身传给SrsListener实例中的(2)

SrsListener* listener = new SrsListener(this, SrsListenerRtmpStream);

listeners.push_back(listener);

//对每一个port建立一个套接字，并新建一个线程。(3)

int port = ::atoi(ports.c_str());

if ((ret = listener->listen(port)) != ERROR_SUCCESS) {

srs_error("RTMP stream listen at port %d failed. ret=%d", port, ret);

return ret;

}

return ret;

}

现在看下SrsListener类中的listen函数实现：

int SrsListener::listen(int port)

{

int ret = ERROR_SUCCESS;

_port = port;

//建立一个TCP套接字

if ((fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {

ret = ERROR_SOCKET_CREATE;

srs_error("create linux socket error. ret=%d", ret);

return ret;

}

srs_verbose("create linux socket success. fd=%d", fd);

int reuse_socket = 1;

if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse_socket, sizeof(int)) == -1) {

ret = ERROR_SOCKET_SETREUSE;

srs_error("setsockopt reuse-addr error. ret=%d", ret);

return ret;

}

srs_verbose("setsockopt reuse-addr success. fd=%d", fd);

sockaddr_in addr;

addr.sin_family = AF_INET;

addr.sin_port = htons(_port);

addr.sin_addr.s_addr = INADDR_ANY;

//绑定端口port ；(4)

if (bind(fd, (const sockaddr*)&addr, sizeof(sockaddr_in)) == -1) {

ret = ERROR_SOCKET_BIND;

srs_error("bind socket error. ret=%d", ret);

return ret;

}

srs_verbose("bind socket success. fd=%d", fd);

//监听端口；(5)

if (::listen(fd, SERVER_LISTEN_BACKLOG) == -1) {

ret = ERROR_SOCKET_LISTEN;

srs_error("listen socket error. ret=%d", ret);

return ret;

}

srs_verbose("listen socket success. fd=%d", fd);

//建立一个_st_netfd_t结构体的实例；(6)

if ((stfd = st_netfd_open_socket(fd)) == NULL){

ret = ERROR_ST_OPEN_SOCKET;

srs_error("st_netfd_open_socket open socket failed. ret=%d", ret);

return ret;

}

srs_verbose("st open socket success. fd=%d", fd);

//SrsListener类中的有一个SrsThread类型的指针，下面标红的代码将开始一个新的线程;(7)

if ((ret = pthread->start()) != ERROR_SUCCESS) {

srs_error("st_thread_create listen thread error. ret=%d", ret);

return ret;

}

srs_verbose("create st listen thread success.");

srs_trace("listen thread cid=%d, current_cid=%d, "

"listen at port=%d, type=%d, fd=%d started success",

pthread->cid(), _srs_context->get_id(), _port, _type, fd);

return ret;

}

现在看下SrsThread类中的start函数：

int SrsThread::start()

{

int ret = ERROR_SUCCESS;

if(tid) {

srs_info("thread already running.");

return ret;

}

//调用了state-threads库函数st_thread_create()，将自身的成员变量thread_fun作为线程运行的主函数，this指针做为thread_fun的一个参数，也就是将SrsThread这个实例传入到thread_fun中；(8)

if((tid = st_thread_create(thread_fun, this, (_joinable? 1:0), 0)) == NULL){

ret = ERROR_ST_CREATE_CYCLE_THREAD;

srs_error("st_thread_create failed. ret=%d", ret);

return ret;

}

// we set to loop to true for thread to run.

loop = true;

// wait for cid to ready, for parent thread to get the cid.

while (_cid < 0 && loop) {

st_usleep(10 * SRS_TIME_MILLISECONDS);

}

// now, cycle thread can run.

can_run = true;

return ret;

}

现在看下SrsThread函数thread_fun：

//此处的arg就是(8)中的this，obj->thread_cycle()即调用的SrsThread实例的thread_cycle(); (9)

void* SrsThread::thread_fun(void* arg)

{

SrsThread* obj = (SrsThread*)arg;

srs_assert(obj);

//obj是什么呢？

obj->thread_cycle();

st_thread_exit(NULL);

return NULL;

}

看下thread_cycle函数

void SrsThread::thread_cycle()

{

int ret = ERROR_SUCCESS;

_srs_context->generate_id();

srs_info("thread cycle start");

_cid = _srs_context->get_id();

srs_assert(handler);

handler->on_thread_start();

// wait for cid to ready, for parent thread to get the cid.

while (!can_run && loop) {

st_usleep(10 * SRS_TIME_MILLISECONDS);

}

while (loop) {

if ((ret = handler->on_before_cycle()) != ERROR_SUCCESS) {

srs_warn("thread on before cycle failed, ignored and retry, ret=%d", ret);

goto failed;

}

srs_info("thread on before cycle success");

//注意标红的代码，这个handler是什么呢？这需要看下SrsThread构造函数对handler是如何初始化的。(10)

if ((ret = handler->cycle()) != ERROR_SUCCESS) {

srs_warn("thread cycle failed, ignored and retry, ret=%d", ret);

goto failed;

}

srs_info("thread cycle success");

if ((ret = handler->on_end_cycle()) != ERROR_SUCCESS) {

srs_warn("thread on end cycle failed, ignored and retry, ret=%d", ret);

goto failed;

}

srs_info("thread on end cycle success");

failed:

if (!loop) {

break;

}

st_usleep(cycle_interval_us);

}

handler->on_thread_stop();

srs_info("thread cycle finished");

}

下面的代码是SrsThread类的构造函数，从中可以看出handler值由thread_handler传递过来的，那么这个thread_handler到底是个什么东西呢？

SrsThread::SrsThread(ISrsThreadHandler* thread_handler, int64_t interval_us, bool joinable)

{

handler = thread_handler;

cycle_interval_us = interval_us;

tid = NULL;

loop = false;

_cid = -1;

_joinable = joinable;

can_run = false;

}

在注释(7)处第一次出现了thread，thread是类SrsListener一个成员，在SysListener构造函数中将listener实例自身传给SrsThread的handler，所以注释(10)中的handler->cycle也就是SrsLintener中的 cycle()函数。

SrsListener::SrsListener(SrsServer* server, SrsListenerType type)

{

fd = -1;

stfd = NULL;

_port = 0;

_server = server;

_type = type;

pthread = new SrsThread(this, 0, true);//注释(11)

}

至此告一段落做个总结，图如下所示：

到最后将是对SrsListener::cycle()的调用。

下一篇：【守望者 rabbitmq 】针对rabbimq 镜像和确认机制的小测试

评论: 0 赞: 0

		自动登录	找回密码
密码			立即注册

[ Ngnix与高并发服务器 ] 【守望者 SRS 】SRS 源代码分析笔记（0.9.194）-分析服务器对端口的监听 ...

推荐阅读