1. Toy模板网首页
  2. > Toy博客

ZLMediaKit源码分析（一）服务启动

9月前 作者：cliffordl 分类：Toy博客 阅读(32) 违法举报

这篇具有很好参考价值的文章主要介绍了ZLMediaKit源码分析（一）服务启动。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

ZLMediaKit源码分析（一）服务启动
ZLMediaKit源码分析（二）推流创建
ZLMediaKit源码分析（三）拉流创建
ZLMediaKit源码分析（四）重封装
ZLMediaKit源码分析（五）代理服务
ZLMediaKit源码分析（六）http 点播

服务创建流程

  1. 创建TcpServer实例，其父类Server同时被实例化。
  2. EventPollerPool初始化线程池。批量创建EventPoller，每个EventPoller对应一个线程，对应一个CPU核。该线程池线程数量对应CPU核数或者由命令行参数-t指定。
  3. 注册TcpServer::_on_create_socket回调，该回调创建Socekt，并且绑定EventPoller。
  4. 启动TcpServer，此时该TcpServer尚未绑定Socket。
  5. 注册TcpServer::_session_alloc回调，该回调函数创建Session，SessionHelper。
    5.1. Session，SessionHelper一一对应，SessionHelper操作全局静态变量session_map，管理session。
    5.2. 注册SessionHelper::_on_create_socket回调，该回调也创建Socket，但是该回调创建的Socket没有Poller负载相关操作，其poller继承自父。
  6. 启动TcpServer，首先TcpServer::setupEvent()创建Socket。
    socket绑定的poller，是在TcpServer的父类Server初始化的时候返回的。
  7. TcpServer::setupEvent()为socket注册两个回调函数。
    7.1. Socket::_on_before_accept，在接受客户端链接请求之前，创建socket，此时绑定的poller有负载考虑。
    7.2. Socket::_on_accept，接受客户端链接请求。使用Socket::_on_before_accept创建的socket，调用步骤4注册的TcpServer::_session_alloc回调创建session。并为该socket绑定数据可读回调。
  8. 启动监听服务。
    8.1. 添加客户端请求事件。
    8.2. 注册客户端请求响应回调，主要调用Socket::onAccept()。
  9. 客户端响应处理。
    9.1. 接受客户端请求accept()。
    9.2. 调用Socket::_on_before_accept回调，创建socket，并考虑到poller负载。
    9.3. 如果8.2. 创建失败，则继承父_poller，创建socket。一般这里不执行。
    9.4. 调用Socket::_on_accept接受客户端请求，核心调用回调TcpServer::onAcceptConnection()，创建客户端session，注册数据可读事件和回调。
    9.5. 注册客户端可读事件，绑定数据可读回调。
  10. 根据poller数量，批量clone服务。

继承关系

ZLMediaKit源码分析（一）服务启动
ZLMediaKit源码分析（一）服务启动

创建TcpServer

创建TcpServer并调用start().

server/main.cpp
        uint16_t rtspPort = mINI::Instance()[Rtsp::kPort];
        auto rtspSrv = std::make_shared<TcpServer>();;
        if (rtspPort) { rtspSrv->start<RtspSession>(rtspPort); }

TcpServer()初始化做了两件重要的事情：

  1. 创建EventPollerPool。
    EventPollerPool是TcpServer父类Server初始化的时候创建的。会根据cpu核数count或者命令行参数-t count创建count个poller实例，并对应启动线程。
  2. 注册Socket创建回调TcpServer::_on_create_socket。该回调有两大用途：
    Server启动时，优先创建socket。
    Server accept客户端链接请求时，创建客户端socket。
3rdpart/ZLToolKit/src/Network/TcpServer.cpp
// main() 函数中 auto rtspSrv = std::make_shared<TcpServer>();;
// poller默认为空
// 初始化父类 Server(poller)
TcpServer::TcpServer(const EventPoller::Ptr &poller) : Server(poller) {
    setOnCreateSocket(nullptr);
}

Server::Server() 初始化

3rdpart/ZLToolKit/src/Network/Server.cpp
Server::Server(EventPoller::Ptr poller) {
	// EventPoller::Ptr Server::_poller;
	// EventPollerPool::Instance() 会创建线程
    _poller = poller ? std::move(poller) : EventPollerPool::Instance().getPoller();
}

创建EventPoller和线程

3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
EventPollerPool::EventPollerPool() {
	// 添加多个EventPoller并创建线程，
    auto size = addPoller("event poller", s_pool_size, ThreadPool::PRIORITY_HIGHEST, true, s_enable_cpu_affinity);
    NoticeCenter::Instance().emitEvent(kOnStarted, *this, size);
    InfoL << "EventPoller created size: " << size;
}

EventPollerPool继承自TaskExecutorGetterImp，addPoller()的实现在父类中。
addPoller会根据cpu核数count或者命令行参数-t count创建count个poller实例，并对应启动线程。

3rdpart/ZLToolKit/src/Thread/TaskExecutor.cpp
size_t TaskExecutorGetterImp::addPoller(const string &name, size_t size, int priority, bool register_thread, bool enable_cpu_affinity) {
    auto cpus = thread::hardware_concurrency();
    size = size > 0 ? size : cpus;
    for (size_t i = 0; i < size; ++i) {
        auto full_name = name + " " + to_string(i);
        EventPoller::Ptr poller(new EventPoller(full_name, (ThreadPool::Priority) priority));
        poller->runLoop(false, register_thread);
        poller->async([i, cpus, full_name, enable_cpu_affinity]() {
            setThreadName(full_name.data());
            if (enable_cpu_affinity) {
                setThreadAffinity(i % cpus);
            }
        });
        // TaskExecutorGetterImp::_threads 成员变量
        // std::vector<TaskExecutor::Ptr> _threads;
        _threads.emplace_back(std::move(poller));
    }
    return size;
}

创建epoll实例。

EventPoller::EventPoller(std::string name, ThreadPool::Priority priority) {
    _name = std::move(name);
    _priority = priority;
    SockUtil::setNoBlocked(_pipe.readFD());
    SockUtil::setNoBlocked(_pipe.writeFD());

#if defined(HAS_EPOLL)
    _epoll_fd = epoll_create(EPOLL_SIZE);
    if (_epoll_fd == -1) {
        throw runtime_error(StrPrinter << "Create epoll fd failed: " << get_uv_errmsg());
    }
    SockUtil::setCloExec(_epoll_fd);
#endif //HAS_EPOLL
    _logger = Logger::Instance().shared_from_this();
    _loop_thread_id = this_thread::get_id();

    //添加内部管道事件
    if (addEvent(_pipe.readFD(), Event_Read, [this](int event) { onPipeEvent(); }) == -1) {
        throw std::runtime_error("Add pipe fd to poller failed");
    }
}

创建线程。

3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
// poller->runLoop(false, register_thread);
void EventPoller::runLoop(bool blocked, bool ref_self) {
    // 默认blocked为false
    if (blocked) {
        ThreadPool::setPriority(_priority);
        lock_guard<mutex> lck(_mtx_running);
        _loop_thread_id = this_thread::get_id();
        if (ref_self) {
            s_current_poller = shared_from_this();
        }
        _sem_run_started.post();
        _exit_flag = false;
        uint64_t minDelay;
#if defined(HAS_EPOLL)
        struct epoll_event events[EPOLL_SIZE];
        while (!_exit_flag) {
            minDelay = getMinDelay();
            startSleep();//用于统计当前线程负载情况
            int ret = epoll_wait(_epoll_fd, events, EPOLL_SIZE, minDelay ? minDelay : -1);
            sleepWakeUp();//用于统计当前线程负载情况
            if (ret <= 0) {
                //超时或被打断
                continue;
            }
            for (int i = 0; i < ret; ++i) {
                struct epoll_event &ev = events[i];
                int fd = ev.data.fd;
                auto it = _event_map.find(fd);
                if (it == _event_map.end()) {
                    epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
                    continue;
                }
                auto cb = it->second;
                try {
                    (*cb)(toPoller(ev.events));
                } catch (std::exception &ex) {
                    ErrorL << "Exception occurred when do event task: " << ex.what();
                }
            }
        }
#else
        // select
        ......
#endif //HAS_EPOLL
    } else {
        _loop_thread = new thread(&EventPoller::runLoop, this, true, ref_self);
        _sem_run_started.wait();
    }
}

根据负载获取EventPoller

声明：

3rdpart/ZLToolKit/src/Poller/EventPoller.h
class EventPollerPool : public std::enable_shared_from_this<EventPollerPool>, public TaskExecutorGetterImp {
public:
	......
    /**
     * 根据负载情况获取轻负载的实例
     * 如果优先返回当前线程，那么会返回当前线程
     * 返回当前线程的目的是为了提高线程安全性
     * @param prefer_current_thread 是否优先获取当前线程
     */
    EventPoller::Ptr getPoller(bool prefer_current_thread = true);
	......
}

实现：

3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
EventPoller::Ptr EventPollerPool::getPoller(bool prefer_current_thread) {
auto poller = EventPoller::getCurrentPoller();
	// prefer_current_thread == true
	// _prefer_current_thread
    if (prefer_current_thread && _prefer_current_thread && poller) {
        return poller;
	}
	// 执行了下面
    return dynamic_pointer_cast<EventPoller>(getExecutor());
}

3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
// static
EventPoller::Ptr EventPoller::getCurrentPoller() {
    return s_current_poller.lock();
}

根据负载返回TaskExecutor。

3rdpart/ZLToolKit/src/Thread/TaskExecutor.cpp
TaskExecutor::Ptr TaskExecutorGetterImp::getExecutor() {
    auto thread_pos = _thread_pos;
    if (thread_pos >= _threads.size()) {
        thread_pos = 0;
    }

    TaskExecutor::Ptr executor_min_load = _threads[thread_pos];
    auto min_load = executor_min_load->load();

    for (size_t i = 0; i < _threads.size(); ++i, ++thread_pos) {
        if (thread_pos >= _threads.size()) {
            thread_pos = 0;
        }

        auto th = _threads[thread_pos];
        auto load = th->load();

        if (load < min_load) {
            min_load = load;
            executor_min_load = th;
        }
        if (min_load == 0) {
            break;
        }
    }
    _thread_pos = thread_pos;
    return executor_min_load;
}

注册TcpServer::_on_create_socket回调

调用：

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
// main() 函数中 auto rtspSrv = std::make_shared<TcpServer>();;
// 所以poller默认为空
// 初始化父类 Server(poller)
TcpServer::TcpServer(const EventPoller::Ptr &poller) : Server(poller) {
    setOnCreateSocket(nullptr);
}

实现：

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
// cb 传入为nullptr
void TcpServer::setOnCreateSocket(Socket::onCreateSocket cb) {
    if (cb) {
        _on_create_socket = std::move(cb);
	} else {
        // 走这里
        // 注册回调函数， 只在这里注册
        // 调用该回调函数时，poller不为空
        // TcpServer::_on_create_socket
        _on_create_socket = [](const EventPoller::Ptr &poller) {
            return Socket::createSocket(poller, false);
        };
    }
	for (auto &pr : _cloned_server) {
        // cb 传入为nullptr
        pr.second->setOnCreateSocket(cb);
    }
}

创建Socket

../3rdpart/ZLToolKit/src/Network/Socket.cpp
Socket::Ptr Socket::createSocket(const EventPoller::Ptr &poller, bool enable_mutex){
    return std::make_shared<Socket>(poller, enable_mutex);
}

如果传入poller参数为空，则寻找最轻负载poller绑定socket。

../3rdpart/ZLToolKit/src/Network/Socket.cpp
Socket::Socket(const EventPoller::Ptr &poller, bool enable_mutex) :
        _mtx_sock_fd(enable_mutex), _mtx_event(enable_mutex),
        _mtx_send_buf_waiting(enable_mutex), _mtx_send_buf_sending(enable_mutex){

    _poller = poller;
    if (!_poller) {
        _poller = EventPollerPool::Instance().getPoller();
    }
    setOnRead(nullptr);
    setOnErr(nullptr);
    setOnAccept(nullptr);
    setOnFlush(nullptr);
    setOnBeforeAccept(nullptr);
    setOnSendResult(nullptr);
}

TcpServer::start()

该函数启动TcpServer服务。主要做两件事情：

  1. 注册TcpServer::_session_alloc回调。
    a) 为SessionType（对应为RtspSession）申请内存空间。绑定socket。
    b) 调用SocketHelper::setOnCreateSocket()，注册创建socket的回调函数，实际调用TcpServer::_on_create_socket。Session无setOnCreateSocket()实现，继承自SocketHelper。
    c) 创建SessionHelper，绑定server和session。Session和SessionHelper一一对应。SessionHelper通过全局变量SessionMap管理session。
  2. 调用start_l(port, host, backlog)启动监听任务。

注册TcpServer::_session_alloc

3rdpart/ZLToolKit/src/Network/TcpServer.h
class TcpServer : public Server {
public:
    using Ptr = std::shared_ptr<TcpServer>;
    ......
	// if (rtspPort) { rtspSrv->start<RtspSession>(rtspPort); }
	// RtspSession等同于SessionType, RtspSession由Session派生
	template<typename SessionType>
    void start(uint16_t port, const std::string &host = "::", uint32_t backlog = 1024) {
        // Session创建器，通过它创建不同类型的服务器
        // 注册创建session的回调
        // std::function<SessionHelper::Ptr(const TcpServer::Ptr &server, const Socket::Ptr &)> TcpServer::_session_alloc;
        _session_alloc = [](const TcpServer::Ptr &server, const Socket::Ptr &sock) {
        	// RtspSession等同于SessionType, RtspSession由Session派生
            auto session = std::make_shared<SessionType>(sock);
         	// 首先调用TcpServer::setOnCreateSocket() 
			// 然后再调用Socket::createSocket()，只注册回调函数
            session->setOnCreateSocket(server->_on_create_socket);
            return std::make_shared<SessionHelper>(server, session);
        };
        start_l(port, host, backlog);
	}
private:
    ......
    Socket::Ptr _socket;
    std::shared_ptr<Timer> _timer;
    Socket::onCreateSocket _on_create_socket;
    std::unordered_map<SessionHelper *, SessionHelper::Ptr> _session_map;
    std::function<SessionHelper::Ptr(const TcpServer::Ptr &server, const Socket::Ptr &)> _session_alloc;
};

RtspSession继承自Session继承自SocketHelper。

3rdpart/ZLToolKit/src/Network/Socket.cpp
void SocketHelper::setOnCreateSocket(Socket::onCreateSocket cb){
	if (cb) {
    	// 走这里，此时cb不为空，TcpServer::_on_create_socket
        _on_create_socket = std::move(cb);
    } else {
        _on_create_socket = [](const EventPoller::Ptr &poller) {
            return Socket::createSocket(poller, false);
        };
    }
}

SocketHelper::_on_create_socket 实际为TcpServer::_on_create_socket，在Socket::onAccept()->Socket::createSocket()中执行。
且当Socket::onAccept()->Socket::_on_before_accept创建失败的时候执行，继承父poller，没有poller负载考虑。

创建SessionHelper管理Session

3rdpart/ZLToolKit/src/Network/Server.cpp
SessionHelper::SessionHelper(const std::weak_ptr<Server> &server, Session::Ptr session) {
    _server = server;
    _session = std::move(session);
	//记录session至全局的map，方便后面管理
	// 这个是个静态变量， 只初始化一次。
    _session_map = SessionMap::Instance().shared_from_this();
    _identifier = _session->getIdentifier();
    _session_map->add(_identifier, _session);
}

TcpServer::start_l()

缩进太深了，专门分析一下。

TcpServer::start_l()

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
void TcpServer::start_l(uint16_t port, const std::string &host, uint32_t backlog) {
	// 1. 创建socket，并绑定回调
	// 创建TCPServer::_socket
	// 注册TCPServer::_socket回调函数setOnBeforeAccept，setOnAccept
	setupEvent();

	// 2. 启动监听
    if (!_socket->listen(port, host.c_str(), backlog)) {
        //创建tcp监听失败，可能是由于端口占用或权限问题
        string err = (StrPrinter << "Listen on " << host << " " << port << " failed: " << get_uv_errmsg(true));
        throw std::runtime_error(err);
    }

    // 3. Timer
    //新建一个定时器定时管理这些tcp会话
    weak_ptr<TcpServer> weak_self = std::dynamic_pointer_cast<TcpServer>(shared_from_this());
    _timer = std::make_shared<Timer>(2.0f, [weak_self]() -> bool {
        auto strong_self = weak_self.lock();
        if (!strong_self) {
            return false;
        }
        strong_self->onManagerSession();
        return true;
    }, _poller);

    // 4. 循环创建多个TcpServer
    EventPollerPool::Instance().for_each([&](const TaskExecutor::Ptr &executor) {
        EventPoller::Ptr poller = dynamic_pointer_cast<EventPoller>(executor);
        if (poller == _poller || !poller) {
            return;
        }
        auto &serverRef = _cloned_server[poller.get()];
        if (!serverRef) {
            serverRef = onCreatServer(poller);
        }
        if (serverRef) {
            serverRef->cloneFrom(*this);
        }
    });

    InfoL << "TCP server listening on [" << host << "]: " << port;
}

TcpServer::setupEvent() 创建监听Socket

setupEvent()有两个主要作用：

  1. 根据poller创建对应的Socket。
  2. 为Socket注册两个回调，在响应客户端请求是调用。
    a) setOnBeforeAccept() 创建客户端socket。根据poller最轻负载创建Socket。
    b) setOnAccept() 接收客户端数据。

此次调用setupEvent()，TcpServer::_poller由TcpServer::TcpServer()->Server::Server()->EventPollerPool::Instance().getPoller(true)创建。

第二次调用setupEvent()在EventPollerPool::Instance().for_each()批量clone的时候，TcpServer::TcpServer()->Server::Server()->EventPollerPool::Instance()->TaskExecutorGetterImp::addPoller()已经批量创建了poller。

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
void TcpServer::setupEvent() {
	// Socket::Ptr TCPServer::_socket;
	// TcpServer::_poller继承父类Server protect变量，此时不为空
    _socket = createSocket(_poller);
weak_ptr<TcpServer> weak_self = std::dynamic_pointer_cast<TcpServer>(shared_from_this());

	// 回调函数1
	// 实际是为Socket::_on_before_accept赋值，回调函数。
	// 该回调函数内调用TcpServer::onBeforeAcceptConnection(poller)
    _socket->setOnBeforeAccept([weak_self](const EventPoller::Ptr &poller) -> Socket::Ptr {
        if (auto strong_self = weak_self.lock()) {
            return strong_self->onBeforeAcceptConnection(poller);
        }
        return nullptr;
	});

	// 回调函数2 
	// 实际是为Socket::_on_accept赋值。
	// 该回调内调用TcpServer::onAcceptConnection()
    _socket->setOnAccept([weak_self](Socket::Ptr &sock, shared_ptr<void> &complete) {
        if (auto strong_self = weak_self.lock()) {
            auto ptr = sock->getPoller().get();
            auto server = strong_self->getServer(ptr);
            ptr->async([server, sock, complete]() {
                //该tcp客户端派发给对应线程的TcpServer服务器
                server->onAcceptConnection(sock);
            });
        }
    });
}

TcpServer创建Socket

  1. 调用TcpServer::createSocket()，
  2. 调用回调函数TcpServer::_on_create_socket(poller)，TcpServer::setOnCreateSocekt()注册该回调为Socket::createSocket()。
  3. 调用Socket::createSocket(), 构造Socket::Socket(const EventPoller::Ptr &poller, bool enable_mutex)，返回Socket::Ptr。

参考：TcpServer::TcpServer(){setOnCreateSocket(nullptr);}

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
// 此时poller不为空，是TcpServer初始化的时候，继承自Server
Socket::Ptr TcpServer::createSocket(const EventPoller::Ptr &poller) {
	// TcpServer::setOnCreateSoket()注册TcpServer::_on_create_socket
	// 实际为Socket::createSocket()，
	return _on_create_socket(poller);
}

注册Socket::setOnBeforeAccept()回调

接收请求之前创建Socket，考虑负载均衡。
调用：

3rdpart/ZLToolKit/src/Network/Server.cpp
void TcpServer::setupEvent() {
	......
	_socket = createSocket(_poller);
	......
	// 回调函数1
	// 实际是为Socket::_on_before_accept赋值，回调函数。
	// 该回调函数内调用TcpServer::onBeforeAcceptConnection(poller)
    _socket->setOnBeforeAccept([weak_self](const EventPoller::Ptr &poller) -> Socket::Ptr {
        if (auto strong_self = weak_self.lock()) {
         	// TcpServer::onBeforeAcceptConnection()
            return strong_self->onBeforeAcceptConnection(poller);
        }
        return nullptr;
	});
	.....
}

为Socket::_on_before_accept赋值：

3rdpart/ZLToolKit/src/Network/Socket.cpp
void Socket::setOnBeforeAccept(onCreateSocket cb){
    LOCK_GUARD(_mtx_event);
	if (cb) {
    	// 走这里
        _on_before_accept = std::move(cb);
    } else {
        _on_before_accept = [](const EventPoller::Ptr &poller) {
            return nullptr;
        };
    }
}

该回调根据poller负载，创建Socket。
参考TcpServer::setupEvent()->TcpServer::createSocket()。
EventPollerPool::Instance().getPoller(false) 会根据poller负载，返回最空闲的poller。
参考：TcpServer::TcpServer()->Server::Server()->EventPollerPool::Instance().getPoller()->TaskExecutorGetterImp::getExecutor()

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
Socket::Ptr TcpServer::onBeforeAcceptConnection(const EventPoller::Ptr &poller) {
    // _poller 只做判断使用
    assert(_poller->isCurrentThread());
    //此处改成自定义获取poller对象，防止负载不均衡
    return createSocket(EventPollerPool::Instance().getPoller(false));
}

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
Socket::Ptr TcpServer::createSocket(const EventPoller::Ptr &poller) {
    return _on_create_socket(poller);
}

注册Socket::setOnAccept()回调

实际上就是为Socket::_on_accept赋值。该回调会在Socket::onAccept()中调用，接受客户端请求。
参数sock由Socket::_on_before_accept()回调创建，该回调根据负载分配一个负载最轻的poller。Socket::_on_before_accept由TcpServer::setupEvent()->Socket::setOnBeforeAccept()设置。
该回调内有重要的函数调用server->onAcceptConnection(sock)，该函数有以下两个主要操作：

  1. 调用TcpServer::_session_alloc创建session，并添加该session至全局静态变量_session_map中。
  2. 为sock设置数据回调，该回调实际调用Session::onRecv()上抛数据。
    调用：
3rdpart/ZLToolKit/src/Network/Server.cpp
void TcpServer::setupEvent() {
	......
	_socket = createSocket(_poller);
	......
	// 回调函数2 
	// 实际是为Socket::_on_accept赋值。
	// 该回调内调用TcpServer::onAcceptConnection()
	// sock实际为Socket::_on_before_accept根据poller负载创建。
    _socket->setOnAccept([weak_self](Socket::Ptr &sock, shared_ptr<void> &complete) {
        if (auto strong_self = weak_self.lock()) {
            auto ptr = sock->getPoller().get();
            // 逆向寻找server
            auto server = strong_self->getServer(ptr);
            ptr->async([server, sock, complete]() {
                //该tcp客户端派发给对应线程的TcpServer服务器
                server->onAcceptConnection(sock);
            });
        }
    });
    ......
}

为Socket::_on_accept赋值：

3rdpart/ZLToolKit/src/Network/Socket.cpp
void Socket::setOnAccept(onAcceptCB cb) {
    LOCK_GUARD(_mtx_event);
	if (cb) {
        // 走这里
        _on_accept = std::move(cb);
    } else {
        _on_accept = [](Socket::Ptr &sock, shared_ptr<void> &complete) {
            WarnL << "Socket not set accept callback, peer fd: " << sock->rawFD();
        };
    }
}
创建并管理Session
3rdpart/ZLToolKit/src/Network/TcpServer.cpp
// Server接收到客户端连接请求，上抛peer Socket至此。
// 一个请求一个Session
Session::Ptr TcpServer::onAcceptConnection(const Socket::Ptr &sock) {
    assert(_poller->isCurrentThread());
    weak_ptr<TcpServer> weak_self = std::dynamic_pointer_cast<TcpServer>(shared_from_this());
	//创建一个Session;这里实现创建不同的服务会话实例
	// SessionHelper
    auto helper = _session_alloc(std::dynamic_pointer_cast<TcpServer>(shared_from_this()), sock);
    auto session = helper->session();
    //把本服务器的配置传递给Session
    session->attachServer(*this);

	//_session_map::emplace肯定能成功
	// 记录 裸指针和智能指针
    auto success = _session_map.emplace(helper.get(), helper).second;
    assert(success == true);

	weak_ptr<Session> weak_session = session;
	// 回调函数1
    //会话接收数据事件
    sock->setOnRead([weak_session](const Buffer::Ptr &buf, struct sockaddr *, int) {
        //获取会话强应用
        auto strong_session = weak_session.lock();
        if (!strong_session) {
            return;
        }
        try {
            strong_session->onRecv(buf);
        } catch (SockException &ex) {
            strong_session->shutdown(ex);
        } catch (exception &ex) {
            strong_session->shutdown(SockException(Err_shutdown, ex.what()));
        }
    });

	SessionHelper *ptr = helper.get();
	// 回调函数2
    //会话接收到错误事件
    sock->setOnErr([weak_self, weak_session, ptr](const SockException &err) {
        //在本函数作用域结束时移除会话对象
        //目的是确保移除会话前执行其onError函数
        //同时避免其onError函数抛异常时没有移除会话对象
        onceToken token(nullptr, [&]() {
            //移除掉会话
            auto strong_self = weak_self.lock();
            if (!strong_self) {
                return;
            }

            assert(strong_self->_poller->isCurrentThread());
            if (!strong_self->_is_on_manager) {
                //该事件不是onManager时触发的，直接操作map
                strong_self->_session_map.erase(ptr);
            } else {
                //遍历map时不能直接删除元素
                strong_self->_poller->async([weak_self, ptr]() {
                    auto strong_self = weak_self.lock();
                    if (strong_self) {
                        strong_self->_session_map.erase(ptr);
                    }
                }, false);
            }
        });

        //获取会话强应用
        auto strong_session = weak_session.lock();
        if (strong_session) {
            //触发onError事件回调
            strong_session->onError(err);
        }
    });
    return session;
}

TcpServer::_session_alloc()在TcpSerer::start()中注册。

3rdpart/ZLToolKit/src/Network/TcpServer.h
class TcpServer : public Server {
public:
    ......
	template<typename SessionType>
    void start(uint16_t port, const std::string &host = "::", uint32_t backlog = 1024) {
        // Session创建器，通过它创建不同类型的服务器
        // 注册创建session的回调
        // std::function<SessionHelper::Ptr(const TcpServer::Ptr &server, const Socket::Ptr &)> TcpServer::_session_alloc;
        _session_alloc = [](const TcpServer::Ptr &server, const Socket::Ptr &sock) {
        	// RtspSession等同于SessionType, RtspSession由Session派生
            auto session = std::make_shared<SessionType>(sock);
         	// 首先调用TcpServer::setOnCreateSocket() 
			// 然后再调用Socket::createSocket()，只注册回调函数
            session->setOnCreateSocket(server->_on_create_socket);
            return std::make_shared<SessionHelper>(server, session);
        };
        start_l(port, host, backlog);
	}
private:
    ......
    std::function<SessionHelper::Ptr(const TcpServer::Ptr &server, const Socket::Ptr &)> _session_alloc;
};

调用TcpServer::_session_alloc()创建Session（包含Socket::_on_before_accept()创建爱你的Socket，其poller负载最轻），并返回SessionHelper。

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
	//创建一个Session;这里实现创建不同的服务会话实例
	// SessionHelper
    auto helper = _session_alloc(std::dynamic_pointer_cast<TcpServer>(shared_from_this()), sock);
    auto session = helper->session();
    //把本服务器的配置传递给Session
    session->attachServer(*this);

3rdpart/ZLToolKit/src/Network/Server.cpp
const Session::Ptr &SessionHelper::session() const {
    return _session;
}

(map<SessionHelper*,SessionHelper::Ptr>) TcpServer::_session_map，记录裸指针和智能指针的对应关系。

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
	//_session_map::emplace肯定能成功
	// 记录 裸指针和智能指针
    auto success = _session_map.emplace(helper.get(), helper).second;
    assert(success == true);
注册Socket::_on_read回调
3rdpart/ZLToolKit/src/Network/TcpServer.cpp
Session::Ptr TcpServer::onAcceptConnection(const Socket::Ptr &sock) {
	省略session创建部分
	......
    //会话接收数据事件
    sock->setOnRead([weak_session](const Buffer::Ptr &buf, struct sockaddr *, int) {
        //获取会话强应用
        auto strong_session = weak_session.lock();
        if (!strong_session) {
            return;
        }
        try {
            strong_session->onRecv(buf);
        } catch (SockException &ex) {
            strong_session->shutdown(ex);
        } catch (exception &ex) {
            strong_session->shutdown(SockException(Err_shutdown, ex.what()));
        }
    });
    ......
}

实现：

3rdpart/ZLToolKit/src/Network/Socket.cpp
void Socket::setOnRead(onReadCB cb) {
    LOCK_GUARD(_mtx_event);
	if (cb) {
        // 走这里
        _on_read = std::move(cb);
    } else {
        _on_read = [](const Buffer::Ptr &buf, struct sockaddr *, int) {
            WarnL << "Socket not set read callback, data ignored: " << buf->size();
        };
    }
}

RtspSession继承自Session。
该回调实际调用纯虚函数Session::onRevc()，实际调用RtspSession::onRecv()。

3rdpart/ZLToolKit/src/Network/Server.h
class Session : public std::enable_shared_from_this<Session>, public SocketHelper {
public:
    using Ptr = std::shared_ptr<Session>;

    Session(const Socket::Ptr &sock);
    ~Session() override = default;

    /**
     * 接收数据入口
     * @param buf 数据，可以重复使用内存区,不可被缓存使用
     */
    virtual void onRecv(const Buffer::Ptr &buf) = 0;
   	......
};

Socket::listen(ip, port)

3rdpart/ZLToolKit/src/Network/Socket.cpp
bool Socket::listen(uint16_t port, const string &local_ip, int backlog) {
    int sock = SockUtil::listen(port, local_ip.data(), backlog);
    if (sock == -1) {
        return false;
    }
    return listen(makeSock(sock, SockNum::Sock_TCP));
}

调用系统接口SockUtil::listen()。

3rdpart/ZLToolKit/src/Network/sckutil.cpp
int SockUtil::listen(const uint16_t port, const char *local_ip, int back_log) {
    int fd = -1;
    int family = support_ipv6() ? (is_ipv4(local_ip) ? AF_INET : AF_INET6) : AF_INET;
    if ((fd = (int)socket(family, SOCK_STREAM, IPPROTO_TCP)) == -1) {
        WarnL << "Create socket failed: " << get_uv_errmsg(true);
        return -1;
    }
    // 设置多路复用
    setReuseable(fd, true, false);
    setNoBlocked(fd);
    setCloExec(fd);

    if (bind_sock(fd, local_ip, port, family) == -1) {
        close(fd);
        return -1;
    }

    //开始监听
    if (::listen(fd, back_log) == -1) {
        WarnL << "Listen socket failed: " << get_uv_errmsg(true);
        close(fd);
        return -1;
    }

    return fd;
}

调用Socket::listen()：

3rdpart/ZLToolKit/src/Network/Socket.cpp
// sock为系统接口socket() 返回套接字ID
bool Socket::listen(const SockFD::Ptr &sock){
    closeSock();
    weak_ptr<SockFD> weak_sock = sock;
    weak_ptr<Socket> weak_self = shared_from_this();
	_enable_recv = true;
	// EventPoller::Ptr Socket::_poller; 
	// epoll实例addEvent绑定套接字
	int result = _poller->addEvent(sock->rawFd(), EventPoller::Event_Read | EventPoller::Event_Error, 
	[weak_self, weak_sock](int event) { // 注册回调函数
        auto strong_self = weak_self.lock();
        auto strong_sock = weak_sock.lock();
        if (!strong_self || !strong_sock) {
            return;
        }
    	// 事件回调创建新链接，这是回调中的具体执行函数。
        strong_self->onAccept(strong_sock, event);
	}
	);

    if (result == -1) {
        return false;
    }

    LOCK_GUARD(_mtx_sock_fd);
    _sock_fd = sock;
    return true;
}

Socket::closeSock()

3rdpart/ZLToolKit/src/Network/Socket.cpp
void Socket::closeSock() {
    _con_timer = nullptr;
    _async_con_cb = nullptr;

    LOCK_GUARD(_mtx_sock_fd);
    _sock_fd  = nullptr;
}

epoll添加请求链接事件

epoll绑定套接字。
PollEventCB注册在Socket::listen(const SockFD::Ptr &sock)。

3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
int EventPoller::addEvent(int fd, int event, PollEventCB cb) {
    TimeTicker();
    if (!cb) {
        WarnL << "PollEventCB is empty";
        return -1;
    }

    if (isCurrentThread()) {
#if defined(HAS_EPOLL)
        struct epoll_event ev = {0};
        ev.events = (toEpoll(event)) | EPOLLEXCLUSIVE;
        ev.data.fd = fd;
        int ret = epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, fd, &ev);
        if (ret == 0) {
            _event_map.emplace(fd, std::make_shared<PollEventCB>(std::move(cb)));
        }
        return ret;
#else
#ifndef _WIN32
        //win32平台，socket套接字不等于文件描述符，所以可能不适用这个限制
        if (fd >= FD_SETSIZE || _event_map.size() >= FD_SETSIZE) {
            WarnL << "select() can not watch fd bigger than " << FD_SETSIZE;
            return -1;
        }
#endif
        auto record = std::make_shared<Poll_Record>();
        record->event = event;
        record->call_back = std::move(cb);
        _event_map.emplace(fd, record);
        return 0;
#endif //HAS_EPOLL
    }

	async([this, fd, event, cb]() {
        // 注册的回调，在Socket::listen(const SockFD::Ptr &sock)中。
        addEvent(fd, event, std::move(const_cast<PollEventCB &>(cb)));
    });
    return 0;
}

同步事件。

3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
// 如果调用该函数的线程就是本对象的轮询线程，那么may_sync为true时就是同步执行任务
// 默认为true
Task::Ptr EventPoller::async(TaskIn task, bool may_sync) {
    return async_l(std::move(task), may_sync, false);
}

3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
Task::Ptr EventPoller::async_l(TaskIn task, bool may_sync, bool first) {
    TimeTicker();
    if (may_sync && isCurrentThread()) {
        task();
        return nullptr;
    }

    auto ret = std::make_shared<Task>(std::move(task));
    {
        lock_guard<mutex> lck(_mtx_task);
        if (first) {
            _list_task.emplace_front(ret);
        } else {
            _list_task.emplace_back(ret);
        }
    }
    //写数据到管道,唤醒主线程
    _pipe.write("", 1);
    return ret;
}

请求响应Socket::onAccept()

接受客户端线程。

  1. accept() 接受客户端请求socket client connect()。
  2. 创建客户端socket，这个地方有两步操作：
    a) 优先执行Socket::_on_before_accept，根据poller负载创建socket。
    b) 保底执行Socket::createSocket(_poller, false)，根据父poller创建socket。
  3. setPeerSock()设置peer socket ip port
  4. 调用Socket::_on_accept创建session，并注册数据上抛回调Socket::_on_read。
    Socket::_on_accept注册在TcpServer::setupEvent()->Socket::setOnAccept()。
  5. Socket::attachEvent() 主要执行以下操作：
    a) 获取共享缓冲区；
    b) 注册数据可读事件；
    c) 数据上抛回调：i.recvFrom()接收数据；ii.执行Socket::onRead()->Socket::_on_read上抛数据。
3rdpart/ZLToolKit/src/Network/Socket.cpp
int Socket::onAccept(const SockFD::Ptr &sock, int event) noexcept {
    int fd;
	while (true) {
        // 读取回调
        if (event & EventPoller::Event_Read) {
            do {
                // 此时只获取客户端fd，没有获取客户端IP
                // connsockfd = accept(listenfd, (struct sockaddr *)&clientaddr, &len);
                fd = (int)accept(sock->rawFd(), nullptr, nullptr);
            } while (-1 == fd && UV_EINTR == get_uv_error(true));

            if (fd == -1) {
                int err = get_uv_error(true);
                if (err == UV_EAGAIN) {
                    //没有新连接
                    return 0;
                }
                auto ex = toSockException(err);
                emitErr(ex);
                ErrorL << "Accept socket failed: " << ex.what();
                return -1;
            }

            SockUtil::setNoSigpipe(fd);
            SockUtil::setNoBlocked(fd);
            SockUtil::setNoDelay(fd);
            SockUtil::setSendBuf(fd);
            SockUtil::setRecvBuf(fd);
            SockUtil::setCloseWait(fd);
            SockUtil::setCloExec(fd);

            // 优先执行这里创建 Socket，考虑负载均衡
            // 只创建Socket使用。
            Socket::Ptr peer_sock;
            try {
                //此处捕获异常，目的是防止socket未accept尽，epoll边沿触发失效的问题
                LOCK_GUARD(_mtx_event);
                //拦截Socket对象的构造
                // 该_poller只用于判断，创建Socket用最轻负载的poller
                peer_sock = _on_before_accept(_poller);
            } catch (std::exception &ex) {
                ErrorL << "Exception occurred when emit on_before_accept: " << ex.what();
                close(fd);
                continue;
            }

            // _on_before_accept(_poller)失败才会走这里
            // 这里继承父poller，没有负载均衡。
            if (!peer_sock) {
                //此处是默认构造行为，也就是子Socket共用父Socket的poll线程并且关闭互斥锁
                peer_sock = Socket::createSocket(_poller, false);
            }

            //设置好fd,以备在onAccept事件中可以正常访问该fd
            // 此时获取客户端IP和port
            // int fd 转换为SockFD::Ptr &peer_sock_fd
            auto peer_sock_fd = peer_sock->setPeerSock(fd);

             // 实际先执行_on_accept(peer_sock, completed);
             // 后执行attachEvent
            shared_ptr<void> completed(nullptr, [peer_sock, peer_sock_fd](void *) {
                try {
                    //然后把该fd加入poll监听(确保先触发onAccept事件然后再触发onRead等事件)
                    if (!peer_sock->attachEvent(peer_sock_fd)) {
                        //加入poll监听失败，触发onErr事件，通知该Socket无效
                        peer_sock->emitErr(SockException(Err_eof, "add event to poller failed when accept a socket"));
                    }
                } catch (std::exception &ex) {
                    ErrorL << "Exception occurred: "<< ex.what();
                }
            });

            try {
                //此处捕获异常，目的是防止socket未accept尽，epoll边沿触发失效的问题
                LOCK_GUARD(_mtx_event);
                //先触发onAccept事件，此时应该监听该Socket的onRead等事件
                // Socket::setOnAccept() 设定的回调函数
                _on_accept(peer_sock, completed);
            } catch (std::exception &ex) {
                ErrorL << "Exception occurred when emit on_accept: " << ex.what();
                continue;
            }
        }  // end if

        if (event & EventPoller::Event_Error) {
            auto ex = getSockErr(sock);
            emitErr(ex);
            ErrorL << "TCP listener occurred a err: " << ex.what();
            return -1;
        }
    }
}
accept()接受客户端请求

系统接口。

创建socket

创建客户端socket，这个地方有两步操作：
a) 优先执行Socket::_on_before_accept，根据poller负载创建socket。
b) 保底执行Socket::createSocket(_poller, false)，根据父poller创建socket。

Socket::_on_before_accept

根据poller负载创建socket。

Socket::createSocket()

Socket::createSocket() 继承父poller创建socket，逻辑上不执行该函数，该函数没有poller负载的考虑。除非Socket::_on_before_accept创建socket失败。
SocketHelper::_on_create_socket 实际调用TcpServer::_on_create_socket。
Socket::_on_before_accept实际也调用的是TcpServer::_on_create_socket。

3rdpart/ZLToolKit/src/Network/Socket.cpp
Socket::Ptr SocketHelper::createSocket(){
    // 使用父_poller，没有负载的考虑
    return _on_create_socket(_poller);
}
设置socket ip port
3rdpart/ZLToolKit/src/Network/Socket.cpp
SockFD::Ptr Socket::setPeerSock(int fd) {
closeSock();
	// 该函数初始化的客户端ip和port
    auto sock = makeSock(fd, SockNum::Sock_TCP);
    LOCK_GUARD(_mtx_sock_fd);
    _sock_fd = sock;
    return sock;
}
Socket::_on_accept接受请求

Socket::_on_accept核心调用TcpServer::onAcceptConnection()：

  1. 创建session；
  2. 注册数据上抛回调Socket::_on_read，实际调用Session::onRecv()->RtspSession::onRecv()。
    赋值：
3rdpart/ZLToolKit/src/Network/Socket.cpp
void Socket::setOnAccept(onAcceptCB cb) {
    LOCK_GUARD(_mtx_event);
	if (cb) {
        // 走了这里
        _on_accept = std::move(cb);
    } else {
        _on_accept = [](Socket::Ptr &sock, shared_ptr<void> &complete) {
            WarnL << "Socket not set accept callback, peer fd: " << sock->rawFD();
        };
    }
}

参考：TcpServer::start_l()->TcpServer::setupEvent()->Socket::setOnAccept()。

3rdpart/ZLToolKit/src/Network/Server.cpp
void TcpServer::setupEvent() {
	......
	_socket = createSocket(_poller);
	......
	// 回调函数2 
	// 实际是为Socket::_on_accept赋值。
	// 该回调内调用TcpServer::onAcceptConnection()
	// sock实际为Socket::_on_before_accept根据poller负载创建。
    _socket->setOnAccept([weak_self](Socket::Ptr &sock, shared_ptr<void> &complete) {
        if (auto strong_self = weak_self.lock()) {
            auto ptr = sock->getPoller().get();
            // 逆向寻找server
            auto server = strong_self->getServer(ptr);
            ptr->async([server, sock, complete]() {
                //该tcp客户端派发给对应线程的TcpServer服务器
                server->onAcceptConnection(sock);
            });
        }
    });
    ......
}
Socket::attachEvent() 绑定数据可读回调

Socket::attachEvent() 主要有以下三个功能：

  1. 获取共享缓冲区；
  2. 注册数据可读事件；
  3. 数据上抛回调；
    i.recvFrom()接收数据；
    ii.执行Socket::onRead()->Socket::_on_read上抛数据。
3rdpart/ZLToolKit/src/Network/Socket.cpp
bool Socket::attachEvent(const SockFD::Ptr &sock) {
    weak_ptr<Socket> weak_self = shared_from_this();
    weak_ptr<SockFD> weak_sock = sock;
    _enable_recv = true;
    _read_buffer = _poller->getSharedBuffer();
    auto is_udp = sock->type() == SockNum::Sock_UDP;
    int result = _poller->addEvent(sock->rawFd(), EventPoller::Event_Read | EventPoller::Event_Error | EventPoller::Event_Write, [weak_self, weak_sock, is_udp](int event) {
        auto strong_self = weak_self.lock();
        auto strong_sock = weak_sock.lock();
        if (!strong_self || !strong_sock) {
            return;
        }

        if (event & EventPoller::Event_Read) {
            strong_self->onRead(strong_sock, is_udp);
        }
        if (event & EventPoller::Event_Write) {
            strong_self->onWriteAble(strong_sock);
        }
        if (event & EventPoller::Event_Error) {
            strong_self->emitErr(getSockErr(strong_sock));
        }
	});

    return -1 != result;
}
申请共享缓冲区。
3rdpart/ZLToolKit/src/Poller/EventPoller.cpp
BufferRaw::Ptr EventPoller::getSharedBuffer() {
    auto ret = _shared_buffer.lock();
    if (!ret) {
        //预留一个字节存放\0结尾符
        ret = BufferRaw::create();
        ret->setCapacity(1 + SOCKET_DEFAULT_BUF_SIZE);
        _shared_buffer = ret;
    }
    return ret;
}
注册数据可读事件

addEvent()数据可读事件。

数据上抛回调

recvFrom()接收数据；执行Socket::onRead()->Socket::_on_read上抛数据。
Socket::onRead() 使用数据上抛回调Socket::_on_read 接收数据并且上抛。

3rdpart/ZLToolKit/src/Network/Socket.cpp
ssize_t Socket::onRead(const SockFD::Ptr &sock, bool is_udp) noexcept{
    ssize_t ret = 0, nread = 0;
    auto sock_fd = sock->rawFd();

    auto data = _read_buffer->data();
    //最后一个字节设置为'\0'
    auto capacity = _read_buffer->getCapacity() - 1;

    struct sockaddr_storage addr;
    socklen_t len = sizeof(addr);

    while (_enable_recv) {
        do {
            nread = recvfrom(sock_fd, data, capacity, 0, (struct sockaddr *)&addr, &len);
        } while (-1 == nread && UV_EINTR == get_uv_error(true));

        if (nread == 0) {
            if (!is_udp) {
                emitErr(SockException(Err_eof, "end of file"));
            } else {
                WarnL << "Recv eof on udp socket[" << sock_fd << "]";
            }
            return ret;
        }

        if (nread == -1) {
            auto err = get_uv_error(true);
            if (err != UV_EAGAIN) {
                if (!is_udp) {
                    emitErr(toSockException(err));
                } else {
                    WarnL << "Recv err on udp socket[" << sock_fd << "]: " << uv_strerror(err);
                }
            }
            return ret;
        }

        if (_enable_speed) {
            // 更新接收速率
            _recv_speed += nread;
        }

        ret += nread;
        data[nread] = '\0';
        //设置buffer有效数据大小
        _read_buffer->setSize(nread);

        //触发回调
        LOCK_GUARD(_mtx_event);
        try {
            //此处捕获异常，目的是防止数据未读尽，epoll边沿触发失效的问题
            // 该回调在Socket::setOnAccept()->TcpServer::onAcceptConnection()->Socket::setOnRead()设置为RtspSession::onRecv()。
            _on_read(_read_buffer, (struct sockaddr *)&addr, len);
        } catch (std::exception &ex) {
            ErrorL << "Exception occurred when emit on_read: " << ex.what();
        }
    }
    return 0;
}

Socket::setOnAccept()->TcpServer::onAcceptConnection()->Socket::setOnRead()注册Socket::_on_read。

3rdpart/ZLToolKit/src/Network/Socket.cpp
void Socket::setOnRead(onReadCB cb) {
    LOCK_GUARD(_mtx_event);
	if (cb) {
        // 走这里
        _on_read = std::move(cb);
    } else {
        _on_read = [](const Buffer::Ptr &buf, struct sockaddr *, int) {
            WarnL << "Socket not set read callback, data ignored: " << buf->size();
        };
    }
}

RtspSession继承自Session。
该回调实际调用纯虚函数Session::onRevc()，实际调用RtspSession::onRecv()。

3rdpart/ZLToolKit/src/Network/Server.h
class Session : public std::enable_shared_from_this<Session>, public SocketHelper {
public:
    using Ptr = std::shared_ptr<Session>;

    Session(const Socket::Ptr &sock);
    ~Session() override = default;

    /**
     * 接收数据入口
     * @param buf 数据，可以重复使用内存区,不可被缓存使用
     */
    virtual void onRecv(const Buffer::Ptr &buf) = 0;
   	......
};

EventPollerPool::Instance().for_each()

循环创建服务器实例

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
void TcpServer::start_l(uint16_t port, const std::string &host, uint32_t backlog) {
    ……

	// for_each()继承自父类
	//一个线程有一个TaskExecutor，现在是用这个函数处理每个 TaskExecutor
	EventPollerPool::Instance().for_each([&](const TaskExecutor::Ptr &executor) {
    	// 如果poller 与当前poller相同，则返回
        EventPoller::Ptr poller = dynamic_pointer_cast<EventPoller>(executor);
        if (poller == _poller || !poller) {
            return;
        }
    	// std::unordered_map<const EventPoller *, Ptr> _cloned_server;
    	// clone (TcpServer)serverRef，相当于有多个server
   	 	// 这里返回serverRef=null
        auto &serverRef = _cloned_server[poller.get()];
     	// 如果serverRef查找失败
		// 则调用onCreatServer()创建TcpServer，并绑定poller。
        if (!serverRef) {
         // 此时poller已创建完毕，poller不为空
            serverRef = onCreatServer(poller);
        }
        if (serverRef) {
            serverRef->cloneFrom(*this);
        }
    });

    InfoL << "TCP server listening on [" << host << "]: " << port;
}

for_each()实现，继承自父类。

3rdpart/ZLToolKit/src/Thread/TaskExecutor.cpp
void TaskExecutorGetterImp::for_each(const function<void(const TaskExecutor::Ptr &)> &cb) {
	for (auto &th : _threads) {
    	// 该回调即为EventPollerPool::Instance().for_each()注册的lambda
        cb(th);
    }
}

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
TcpServer::Ptr TcpServer::onCreatServer(const EventPoller::Ptr &poller) {
    return std::make_shared<TcpServer>(poller);
}

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
void TcpServer::cloneFrom(const TcpServer &that) {
    if (!that._socket) {
        throw std::invalid_argument("TcpServer::cloneFrom other with null socket");
	}
	// 创建Socket::Ptr TcpServer::_socket
    setupEvent();
    _on_create_socket = that._on_create_socket;
	_session_alloc = that._session_alloc;
	// 克隆一个相同fd的Socket对象
	// 拷贝socket 包含SockFD，并listen。
    _socket->cloneFromListenSocket(*(that._socket));
    weak_ptr<TcpServer> weak_self = std::dynamic_pointer_cast<TcpServer>(shared_from_this());
    _timer = std::make_shared<Timer>(2.0f, [weak_self]() -> bool {
        auto strong_self = weak_self.lock();
        if (!strong_self) {
            return false;
        }
        strong_self->onManagerSession();
        return true;
    }, _poller);
    this->mINI::operator=(that);
    _parent = &that;
}

setupEvent() 参考：TcpServer::start()->TcpServer::start_l()->TcpServer::setupEvent()。

3rdpart/ZLToolKit/src/Network/Socket.cpp
bool Socket::cloneFromListenSocket(const Socket &other) {
    auto sock = cloneSockFD(other);
    if (!sock) {
        return false;
    }
    return listen(sock);
}

3rdpart/ZLToolKit/src/Network/Socket.cpp
SockFD::Ptr Socket::cloneSockFD(const Socket &other) {
    SockFD::Ptr sock;
    {
        LOCK_GUARD(other._mtx_sock_fd);
        if (!other._sock_fd) {
            WarnL << "sockfd of src socket is null";
            return nullptr;
        }
        sock = std::make_shared<SockFD>(*(other._sock_fd), _poller);
    }
    return sock;
}

3rdpart/ZLToolKit/src/Network/TcpServer.cpp
void TcpServer::onManagerSession() {
    assert(_poller->isCurrentThread());

    onceToken token([&]() {
        _is_on_manager = true;
    }, [&]() {
        _is_on_manager = false;
    });

    // std::unordered_map<SessionHelper *, SessionHelper::Ptr> _session_map;
    for (auto &pr : _session_map) {
        //遍历时，可能触发onErr事件(也会操作_session_map)
        try {
            pr.second->session()->onManager();
        } catch (exception &ex) {
            WarnL << ex.what();
        }
    }
}

–END–
参考：
https://blog.csdn.net/yudaichenydc/article/details/127997301文章来源地址https://www.toymoban.com/news/detail-508841.html

到了这里，关于ZLMediaKit源码分析（一）服务启动的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

  • ZLMediaKit(webrtc)在linux上（CentOS7）部署与启动

    一.ZLMediaKit(webrtc)在CentOS7部署与启动 步骤1：安装ffmpeg 步骤2：gcc安装 步骤3：cmake安装 version `GLIBCXX_3.4.20‘ not found 解决方法 步骤4：openssl 安装 (openssl版本要求1.1以上) 步骤5：安装 yasm 步骤6：libsrtp安装 步骤7：下载zlm源码 编译 修改配置文件 运行 测试webrtc:

    2024年02月13日
    浏览(51)
  • SpringBoot源码分析：SpringBoot启动源码(一)

    SpringBoot源码分析：SpringBoot启动源码(一)

    SpringBoot启动的源码分为前期初始化，和后期启动两个部分，我们从这两个部分开始介绍。 SpringBoot前期初始化主要由下面三行代码组成。  进入SpringApplication.getSpringFactoriesInstances方法  最终进入SpringFactoriesLoader.loadFactoryNames方法  最后SpringFactoriesLoader.loadSpringFactories会到sprin

    2024年02月07日
    浏览(42)

  • flink源码分析 - flink命令启动分析

    flink版本: flink-1.12.1 源码位置:  flink-dist/src/main/flink-bin/bin/flink flink命令源码: 首先讲第一段: 工作中，很多人喜欢用符号链接(软连接)去将原始命令链接到一个新的文件。 例如:  将 /home/aaa链接到/opt/soft/flink-1.12.1/bin/flink，  实际使用的时候就可以用 aaa去代替flink命令。 例如

    2024年01月18日
    浏览(38)
  • 【框架源码】SpringBoot核心源码解读之启动类源码分析

    【框架源码】SpringBoot核心源码解读之启动类源码分析

    首先我们要先带着我们的疑问，spring boot是如何启动应用程序？去分析SpringBoot的启动源码。 我们在新建SpringBoot项目时，核心方法就是主类的run方法。 SpringApplication.run(ArchWebApplication.class, args) 我们点击run方法进入到源码中，这块传入的了一个我们当前程序主类的类对象以及主

    2024年02月06日
    浏览(42)
  • SpringBoot的启动流程源码分析

    SpringBoot的启动流程源码分析

    new 一个IOC容器，传入配置好的文件xml,在这个地方打bug 在这个debug的栈帧中，下面几个不用看，直接看到getBean 内容如图所示，name传的就是我们在xml的bean标签的id,这里是instanceA 进入到doGetBean后，因为我是从IOC初始化容器debug进来的所以第一次通过。 Object sharedInstance = getSingl

    2024年01月24日
    浏览(47)
  • Spring Boot启动源码分析

    Spring Boot启动源码分析

    版本：spring-boot-starter-parent版本为2.3.0 Spring Boot项目的启动入口是一个main方法，因此我们从该方法入手即可 跟踪run方法 这里分两步debug: new SpringApplication( primarySources ) 执行run()方法 deduceFromClasspath推断应用程序类型 该方法根据是否存在指定路径的类来推断应用程序类型。有

    2024年02月07日
    浏览(47)

  • 基于aarch64分析kernel源码 三：启动代码分析

    这段代码是主要的入口点函数 primary_entry ，在 启动过程 中执行 一系列操作 后 将控制权切换到内核 。 以下是对每行代码的解释： 这行代码表示这是一个 本地符号 ，标识 primary_entry 函数的 开始位置 。 这三行代码调用了三个不同的函数。首先调用 record_mmu_state 函数来 记录

    2024年02月15日
    浏览(44)
  • SpringBoot源码学习4——SpringBoot内嵌Tomcat启动流程源码分析

    SpringBoot源码学习4——SpringBoot内嵌Tomcat启动流程源码分析

    系列文章目录和关于我 我在初学spring的时候，很懵逼，因为整个项目中不存在main方法，让我有点摸不着头脑。那时候我知道有个东西叫tomcat是它监听了端口，解析了协议调到了我的servlet。 在我初学SpringBoot的时候，很懵逼，有main方法了，但是tomcat在哪里呢，又是如何启动起

    2024年02月04日
    浏览(46)
  • Activity启动过程详解(Android 12源码分析)

    Activity启动过程详解(Android 12源码分析)

    启动一个Activity，通常有两种情况，一种是在应用内部启动Activity，另一种是Launcher启动 1、应用内启动 通过startActivity来启动Activity 启动流程： 一、Activity启动的发起 二、Activity的管理——ATMS 三、线程切换即消息处理——mH 四、Activity启动核心实现——初始化及生命周期 2、

    2024年02月13日
    浏览(46)
  • SpringBoot3.X源码分析(启动流程)

    SpringBoot3.X源码分析(启动流程)

    1 启动入口 静态辅助类，可用于运行使用默认配置（即我们添加的一系列注解）的指定源的 SpringApplication 。 primarySource - 要载入的主要源，即指定源，这里为传入的 Application.class  Class? ：泛型决定了任何类都可以传入 args - 应用程序参数（通常从main方法传递） 返回：正在运

    2024年01月23日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包