目录1 runLoop方法2 管道机制2.1 管道与epoll的结合2.2 管道封装PipeWrap类2.3 管道机制实现3 完整对象图4 线程任务功能实现流程图5 对外接口6 职责总结7 与EventPollerPool类的关系EventPoller类就是干活的牛马——它是个线程类。1 runLoop方法EventPoller类的runLoop方法是一个对线程的抽象、线程运行的实体所在的方法——这个方法一物两用它既是线程运行入口也是线程创建的代码所在地。该方法的if分支是该线程运行实体为何放在if里因为程序的局部性原理要把大概率会执行的代码放到if分支下因为这样提高cache命中率else分支是创建该线程的代码所在。其实现概略如下voidEventPoller::runLoop(boolblocked,boolref_self){if(blocked){_sem_run_started.post();_exit_flagfalse;uint64_tminDelay;#ifdefined(HAS_EPOLL)structepoll_eventevents[EPOLL_SIZE];while(!_exit_flag){minDelaygetMinDelay();startSleep();//用于统计当前线程负载情况intretepoll_wait(_event_fd,events,EPOLL_SIZE,minDelay?minDelay:-1);sleepWakeUp();//用于统计当前线程负载情况if(ret0){//超时或被打断 [AUTO-TRANSLATED:7005fded]//Timed out or interruptedcontinue;}_event_cache_expired.clear();for(inti0;iret;i){structepoll_eventevevents[i];intfdev.data.fd;if(_event_cache_expired.count(fd)){//event cache refreshcontinue;}autoit_event_map.find(fd);if(it_event_map.end()){epoll_ctl(_event_fd,EPOLL_CTL_DEL,fd,nullptr);continue;}autocbit-second;try{(*cb)(toPoller(ev.events));}catch(std::exceptionex){ErrorLException occurred when do event task: ex.what();}}}#elifdefined(HAS_KQUEUE)structkeventkevents[KEVENT_SIZE];while(!_exit_flag){minDelaygetMinDelay();structtimespectimeout{(long)minDelay/1000,(long)minDelay%1000*1000000};startSleep();intretkevent(_event_fd,nullptr,0,kevents,KEVENT_SIZE,minDelay?timeout:nullptr);sleepWakeUp();if(ret0){continue;}…………}#elseintret,max_fd;FdSet set_read,set_write,set_err;ListPoll_Record::Ptrcallback_list;structtimevaltv;while(!_exit_flag){//定时器事件中可能操作_event_map [AUTO-TRANSLATED:f2a50ee2]//Possible operations on _event_map in timer eventsminDelaygetMinDelay();tv.tv_sec(decltype(tv.tv_sec))(minDelay/1000);tv.tv_usec1000*(minDelay%1000);set_read.fdZero();set_write.fdZero();set_err.fdZero();max_fd0;for(autopr:_event_map){if(pr.firstmax_fd){max_fdpr.first;}if(pr.second-eventEvent_Read){set_read.fdSet(pr.first);//监听管道可读事件}if(pr.second-eventEvent_Write){set_write.fdSet(pr.first);//监听管道可写事件}if(pr.second-eventEvent_Error){set_err.fdSet(pr.first);//监听管道错误事件}}startSleep();//用于统计当前线程负载情况retzl_select(max_fd1,set_read,set_write,set_err,minDelay?tv:nullptr);sleepWakeUp();//用于统计当前线程负载情况…………}#endif//HAS_EPOLL}else{_loop_threadnewthread(EventPoller::runLoop,this,true,ref_self);_sem_run_started.wait();}}可以看到它支持跨平台不管linux还是windows等有不同实现。linux下自然是epoll调用苹果os是kevent其他平台则是自己实现的一套epoll类似机制。本篇关注linux下实现if线程运行的实体主要是采用了epoll机制核心是进行epoll_wait调用然后回调对应fd的方法。至于超时时间是结合了定期器任务队列来定的。它的职责就是处理延时任务、定时任务和立即执行的任务。这一切任务的投递基础是管道投递任务。管道机制怎么实现的2 管道机制如何给这个牛马分配任务让它干活呢通过管道。管道是它的组成部分它们是组合关系。可以看PipeWrap见名知意管道封装还能看到到eventpoller的epoll的fd——_event_fd以及两个对外接口async和doDelayTask。2.1 管道与epoll的结合linux平台下不管是管道还是socket等等都抽象为fd文件描述符所以epoll_wait可以同时监听管道或者socket。在eventpoller对象创建时构造函数里创建了epoll对象然后也创建了管道对象然后把管道fd加入到该epoll中监听其可读事件固定绑定管道可读事件的回调统一为onPipeEvent()实现参见2.3里面就是把投递的任务遍历执行。如何投递任务?async接口中先是把任务lambda表达式闭包然后通过async_l加入到_list_task的vector中然后_pipe.write(“”, 1);这句只为了唤醒线程runLoop里的epoll_wait执行管道的读回调——前面说了统一为onPipeEvent()然后遍历_list_task执行所有投递任务。注意:这些操作是对eventpoller对象上进行的操作它提供的能力。因此所有投递的任务都是以管道为基础这样通信的最底层统一了。简单就是稳定。2.2 管道封装PipeWrap类管道的封装抽象出了PipeWrap类namespacetoolkit{classPipeWrap{public:PipeWrap();~PipeWrap();intwrite(constvoid*buf,intn);intread(void*buf,intn);intreadFD()const{return_pipe_fd[0];}intwriteFD()const{return_pipe_fd[1];}voidreOpen();private:voidclearFD();private:int_pipe_fd[2]{-1,-1};};}/* namespace toolkit */namespacetoolkit{PipeWrap::PipeWrap(){reOpen();}voidPipeWrap::reOpen(){clearFD();#ifdefined(_WIN32)constchar*localipSockUtil::support_ipv6()?::1:127.0.0.1;autolistener_fdSockUtil::listen(0,localip);checkFD(listener_fd)SockUtil::setNoBlocked(listener_fd,false);autolocalPortSockUtil::get_local_port(listener_fd);_pipe_fd[1]SockUtil::connect(localip,localPort,false);checkFD(_pipe_fd[1])_pipe_fd[0](int)accept(listener_fd,nullptr,nullptr);checkFD(_pipe_fd[0])SockUtil::setNoDelay(_pipe_fd[0]);SockUtil::setNoDelay(_pipe_fd[1]);close(listener_fd);#elseif(pipe(_pipe_fd)-1){throwruntime_error(StrPrinterCreate posix pipe failed: get_uv_errmsg());}#endif// defined(_WIN32)SockUtil::setNoBlocked(_pipe_fd[0],true);SockUtil::setNoBlocked(_pipe_fd[1],false);SockUtil::setCloExec(_pipe_fd[0]);SockUtil::setCloExec(_pipe_fd[1]);}windows下管道是对本地socket的封装其他oslinux等就是pipe调用。以linux为例pipe()会在内核中开辟一段缓冲区并返回一对文件描述符形成一个 单向数据通道 int_pipe_fd[2];if(pipe(_pipe_fd)-1){perror(pipe);exit(EXIT_FAILURE);}成功返回0失败返回-1_pipe_fd[0]→ 读端 从这读数据_pipe_fd[1]→ 写端 往这写数据数据流向就像水管写入_pipe_fd[1]的字节能从_pipe_fd[0]读出来 先写先读FIFO 。这两个描述符可以像普通文件一样用read()/write()/close()操作。Linux 上其实有更轻量的eventfd()一个 8 字节计数的 fd专为这事设计但它是 Linux 独有 的。ZLMediaKit 要跑在 macOSkqueue、BSD 甚至 Windows 上所以选了 POSIX 通用的pipe()作为最大公约数。这是跨平台项目里很典型的取舍 放弃单平台最优解换全平台一致性 。2.3 管道机制实现固定绑定管道可读事件的回调统一为onPipeEvent()是如何实现的EventPoller::EventPoller(std::string name){#ifdefined(HAS_EPOLL)||defined(HAS_KQUEUE)_event_fdcreate_event();if(_event_fd-1){throwruntime_error(StrPrinterCreate event fd failed: get_uv_errmsg());}SockUtil::setCloExec(_event_fd);#endif//HAS_EPOLL_namestd::move(name);_loggerLogger::Instance().shared_from_this();addEventPipe();}voidEventPoller::addEventPipe(){SockUtil::setNoBlocked(_pipe.readFD());SockUtil::setNoBlocked(_pipe.writeFD());// 添加内部管道事件 [AUTO-TRANSLATED:6a72e39a]//Add internal pipe eventif(addEvent(_pipe.readFD(),EventPoller::Event_Read,[this](intevent){onPipeEvent();})-1){throwstd::runtime_error(Add pipe fd to poller failed);}}EventPoller的构造函数EventPoller::EventPoller调用EventPoller::addEventPipe里再调用addEvent进行绑定管道可读回调onPipeEvent()的。intEventPoller::addEvent(intfd,intevent,PollEventCB cb){TimeTicker();if(!cb){WarnLPollEventCB is empty;return-1;}if(isCurrentThread()){#ifdefined(HAS_EPOLL)structepoll_eventev{0};ev.eventstoEpoll(event);ev.data.fdfd;intretepoll_ctl(_event_fd,EPOLL_CTL_ADD,fd,ev);if(ret!-1){_event_map.emplace(fd,std::make_sharedPollEventCB(std::move(cb)));}returnret;#elifdefined(HAS_KQUEUE)structkeventkev[2];intindex0;if(eventEvent_Read){EV_SET(kev[index],fd,EVFILT_READ,EV_ADD|EV_CLEAR,0,0,nullptr);}if(eventEvent_Write){EV_SET(kev[index],fd,EVFILT_WRITE,EV_ADD|EV_CLEAR,0,0,nullptr);}intretkevent(_event_fd,kev,index,nullptr,0,nullptr);if(ret!-1){_event_map.emplace(fd,std::make_sharedPollEventCB(std::move(cb)));}returnret;#else#ifndef_WIN32// win32平台socket套接字不等于文件描述符所以可能不适用这个限制 [AUTO-TRANSLATED:6adfc664]//On the win32 platform, the socket does not equal the file descriptor, so this restriction may not applyif(fdFD_SETSIZE){WarnLselect() can not watch fd bigger than FD_SETSIZE;return-1;}#endifautorecordstd::make_sharedPoll_Record();record-fdfd;record-eventevent;record-call_backstd::move(cb);_event_map.emplace(fd,record);return0;#endif}async([this,fd,event,cb]()mutable{addEvent(fd,event,std::move(cb));});return0;}EventPoller::addEvent实现了跨平台的事件监听如第1节所述linux下自然是epoll调用苹果os是kevent其他平台则是自己实现的一套epoll。linux就是通过epoll_ctl把管道的可读fd加入监听后在第1节里的EventPoller::runLoop里通过epoll_wait进行监听。还有个关键容器_event_map它把fd和其回调进行了统一管理这样在epoll_wait唤醒后拿着fd就可以从_event_map查找到对应的回调函数。那么怎么触发管道可读呢如下实现Task::PtrEventPoller::async_l(TaskIn task,boolmay_sync,boolfirst){TimeTicker();if(may_syncisCurrentThread()){task();returnnullptr;}autoretstd::make_sharedTask(std::move(task));{lock_guardmutexlck(_mtx_task);if(first){_list_task.emplace_front(ret);}else{_list_task.emplace_back(ret);}}//写数据到管道,唤醒主线程 [AUTO-TRANSLATED:2ead8182]//Write data to the pipe and wake up the main thread_pipe.write(,1);returnret;}可以看到 _pipe.write(“”, 1); 就是触发这就是管道机制。管道机制只是为了唤醒线程真正干活的就是它承载的任务回调函数。3 完整对象图4 线程任务功能实现流程图简略流程图如上其工作基本就是这么简单的流程。上图都是EventPoller的方法只不过runLoop是线程函数入口是其工作核心。async和doDelayTask是对外的主要接口。因为async_first、async_l都是内部方法——被对外方法调用的。5 对外接口只关注这些接口async_first、async、async_l和doDelayTask。由第3节图可知核心的对外接口是两个:async和doDelayTask async_first和async都是对async_l的调用。Task::PtrEventPoller::async(TaskIn task,boolmay_sync){returnasync_l(std::move(task),may_sync,false);}Task::PtrEventPoller::async_first(TaskIn task,boolmay_sync){returnasync_l(std::move(task),may_sync,true);}EventPoller::DelayTask::PtrEventPoller::doDelayTask(uint64_tdelay_ms,functionuint64_t()task){DelayTask::Ptr retstd::make_sharedDelayTask(std::move(task));autotime_linegetCurrentMillisecond()delay_ms;async_first([time_line,ret,this](){//异步执行的目的是刷新select或epoll的休眠时间 [AUTO-TRANSLATED:a6b5c8d7]//The purpose of asynchronous execution is to refresh the sleep time of select or epoll_delay_task_map.emplace(time_line,ret);});returnret;}可以看到async接口调用async_l接口通知将封包的task插入队列_list_task尾部让EventPoller的runLoop线程按照先来后到进行处理。async_first接口调用async_l接口通知将封包的task插入队列_list_task头部让EventPoller的runLoop线程优先处理。doDelayTask接口呢先封装闭包回调加入队列_delay_task_map然后调用async_first让runLoop线程优先处理——当runLoop线程执行pipe通道的读事件遍历队列时首先执行这个闭包因为放到了对头——加入队列_delay_task_map。而runLoop线程在epollwait之前会处理定时器到期任务。6 职责总结它的主要能力核心构成:第一个是它是线程函数实体所在第二线程里epoll的机制。epoll天生用来监听fd的soket管道都是fd天然支持所以形成了管道这个投递任务的基础核心机制。基于管道任投递任务的机制衍生出了定时任务投递。所以其职责也就很清楚了处理各种任务如下:事件监听与分发通过 epoll 等机制监听文件描述符上的事件如套接字的可读、可写事件等将这些事件分发给相应的回调函数进行处理实现对网络I/O事件的高效处理。​定时任务管理负责管理延迟任务和定时任务。通过 doDelayTask 函数可以添加延迟执行的任务任务会根据设定的延迟时间被插入到任务队列中并在到期时由 flushDelayTasks 函数执行。对于循环任务还会在执行后根据设定重新插入队列。​线程间通信与同步通过管道等方式实现线程间的通信和同步。例如当有新的任务需要添加或其他线程需要通知 EventPoller 进行某些操作时可以通过往管道写数据来唤醒 EventPoller 的事件循环线程使其及时处理相关任务。​资源管理与释放在事件循环过程中对相关资源进行管理如文件描述符的注册与注销、内存的分配与释放等确保资源的正确使用和及时释放避免资源泄漏。7 与EventPollerPool类的关系EventPollerPool管理EventPollerEventPoller是线程那么EventPollerPool就是线程池它们是聚合关系或者组合关系。1个EventPollerPool包含n个EventPoller对象。其关系图如下当实例化EventPollerPool时自会实例化n个EventPoller启动各个EventPoller的runLoop线程。