#muduo学习笔记(二)Reactor关键结构

目录 muduo学习笔记(二)Reactor关键结构 Reactor简述 什么是Reactor Reactor模型的优缺点 poll简述 poll使用样例 muduo Reactor关键结构 Channel Poller EventLoop Reactor时序图 测试程序-单次触发的定时器 muduo学习笔记(二)Reactor关键结构 Reactor简述 什么是Reactor Reactor是一种基于事件驱动的设计模式,即通过回调机制,我们将事件的接口注册到Reactor上,当事件发生之后,就会回调注册的接口。 Reactor必要的几个组件: Event Multiplexer事件分发器:即一些I/O复用机制select、poll、epoll等.程序将事件源注册到分发器上,等待事件的触发,做相应处理. Handle事件源:用于标识一个事件,Linux上是文件描述符. Reactor反应器:用于管理事件的调度及注册删除.当有激活的事件时,则调用回调函数处理,没有则继续事件循环. event handler事件处理器:管理已注册事件和的调度,分成不同类型的事件(读/写,定时)当事件发生,调用对应的回调函数处理. Reactor模型的优缺点 优点 1)响应快,不必为单个同步时间所阻塞,虽然Reactor本身依然是同步的; 2)编程相对简单,可以最大程度的避免复杂的多线程及同步问题,并且避免了多线程/进程的切换开销; 3)可扩展性,可以方便的通过增加Reactor实例个数来充分利用CPU资源; 4)可复用性,reactor框架本身与具体事件处理逻辑无关,具有很高的复用性; 缺点 Reactor模式在IO读写数据时还是在同一个线程中实现的,即使使用多个Reactor机制的情况下,那些共享一个Reactor的Channel如果出现一个长时间的数据读写,会影响这个Reactor中其他Channel的相应时间,比如在大文件传输时,IO操作就会影响其他Client的相应时间,因而对这种操作,使用传统的Thread-Per-Connection或许是一个更好的选择,或则此时使用Proactor模式。 poll简述 poll的使用方法与select相似,轮询多个文件描述符,有读写时设置相应的状态位,poll相比select优在没有最大文件描述符数量的限制. # include int poll ( struct pollfd * fds, unsigned int nfds, int timeout); struct pollfd { int fd; /* 文件描述符 */ short events; /* 等待的事件 */ short revents; /* 实际发生了的事件 */ } ;   每一个pollfd结构体指定了一个被监视的文件描述符,可以传递多个结构体,指示poll()监视多个文件描述符。每个结构体的events域是监视该文件描述符的事件掩码,由用户来设置这个域。revents域是文件描述符的操作结果事件掩码,内核在调用返回时设置这个域。events域中请求的任何事件都可能在revents域中返回。合法的事件如下:   POLLIN         有数据可读。   POLLRDNORM      有普通数据可读。   POLLRDBAND      有优先数据可读。   POLLPRI         有紧迫数据可读。   POLLOUT       写数据不会导致阻塞。   POLLWRNORM      写普通数据不会导致阻塞。   POLLWRBAND      写优先数据不会导致阻塞。   POLLMSGSIGPOLL     消息可用。 poll使用样例 #include #include #include #include #include #include #include #include #define MAX_BUFFER_SIZE 1024 #define IN_FILES 1 #define MAX(a,b) ((a>b)?(a):(b)) int main(int argc ,char **argv) { struct pollfd fds[3]; char buf[1024]; int i,res,real_read, maxfd; if((fds[0].fd=open("/dev/stdin",O_RDONLY|O_NONBLOCK)) < 0) { fprintf(stderr,"open data1 error:%s",strerror(errno)); return 1; } for (i = 0; i < IN_FILES; i++) { fds[i].events = POLLIN | POLLPRI; } while(1) //|| fds[1].events || fds[2].events) { int ret = poll(fds, 1, 1000); if (ret < 0) { printf("Poll error : %s\n",strerror(errno)); return 1; } if(ret == 0){ printf("Poll timeout\n"); continue; } for (i = 0; i< 1; i++) { if (fds[i].revents) { memset(buf, 0, MAX_BUFFER_SIZE); real_read = read(fds[i].fd, buf, MAX_BUFFER_SIZE); if (real_read < 0) { if (errno != EAGAIN) { printf("read eror : %s\n",strerror(errno)); continue; } } else if (!real_read) { close(fds[i].fd); fds[i].events = 0; } else { if (i == 0) { buf[real_read] = '\0'; printf("%s", buf); if ((buf[0] == 'q') || (buf[0] == 'Q')) { printf("quit\n"); return 1; } } else { buf[real_read] = '\0'; printf("%s", buf); } } } } } exit(0); } muduo Reactor关键结构 muduo Reactor最核心的事件分发机制, 即将IO multiplexing拿到的IO事件分发给各个文件描述符(fd)的事件处理函数。 Channel Chanel目前我对它的理解是,它负责管理一个文件描述符(file descript)IO事件. Channel会封装C的poll,把不同的IO事件分发到不同的回调:ReadCallBack、WriteCallBack等 每个Channel对象自始至终只属于一个EventLoop,因此每个Channel对象都只属于某一个IO线程。 每个Channel对象自始至终只负责一个文件描述符(fd) 的IO事件分发 #ifndef NET_CHANNEL_H #define NET_CHANNEL_H #include #include "EventLoop.hh" class Channel { public: typedef std::function EventCallBack; Channel(EventLoop* loop, int fd); ~Channel(); void handleEvent(); void setReadCallBack(const EventCallBack& cb) { m_readCallBack = cb; } void setWriteCallBack(const EventCallBack& cb) { m_writeCallBack = cb; } void setErrorCallBack(const EventCallBack& cb) { m_errorCallBack = cb; } int fd() const { return m_fd; } int events() const { return m_events; } void set_revents(int revt) { m_revents = revt; } bool isNoneEvent() const { return m_events == kNoneEvent; } void eableReading() { m_events |= kReadEvent; update(); } int index() { return m_index; } void set_index(int idx) { m_index =idx; } EventLoop* ownerLoop() { return m_pLoop; } private: Channel& operator=(const Channel&); Channel(const Channel&); void update(); static const int kNoneEvent; static const int kReadEvent; static const int kWriteEvent; EventLoop* m_pLoop; const int m_fd; int m_events; // 等待的事件 int m_revents; // 实际发生了的事件 int m_index; EventCallBack m_readCallBack; EventCallBack m_writeCallBack; EventCallBack m_errorCallBack; }; #endif //Channel.cpp #include #include "Channel.hh" #include "Logger.hh" const int Channel::kNoneEvent = 0; const int Channel::kReadEvent = POLLIN | POLLPRI; const int Channel::kWriteEvent = POLLOUT; Channel::Channel(EventLoop* loop, int fd) : m_pLoop(loop), m_fd(fd), m_events(0), m_revents(0), m_index(-1) { } Channel::~Channel() { } void Channel::update() { m_pLoop->updateChannel(this); } void Channel::handleEvent() { if(m_revents & POLLNVAL) { LOG_WARN << "Channel::handleEvent() POLLNVAL"; } if(m_revents & (POLLERR | POLLNVAL)){ if(m_errorCallBack) m_errorCallBack(); } if(m_revents & (POLLIN | POLLPRI | POLLRDHUP)){ if(m_readCallBack) m_readCallBack(); } if(m_revents & POLLOUT){ if(m_writeCallBack) m_writeCallBack(); } } 值得一提的就是 Channel::update()它会调用EventLoop::updateChannel(), 后者会转而调 用Poller::updateChannel()。Poller对象下面会讲,通过Poller::updateChannel()注册IO事件(即file descript). Channel::handleEvent()是Channel的核心, 它由EventLoop::loop()调 用, 它的功能是根据revents发生事件的的值分别调用不同的用户回调。 这个函数以后还会扩充。 Poller Poller class是IO multiplexing的封装。 它现在是个具体类,而在muduo中是个抽象基类,因为muduo同时支持poll(2)和epoll(4)两种IOmultiplexing机制。 Poller是EventLoop的间接成员,只供其自己在EventLoop的IO线程中调用,因此无须加锁。其生命期与EventLoop相等。 Poller并不拥有管理文件描述符事件的Channel, Channel在析构之前必须自己 unregister(EventLoop::removeChannel()) , 避免空悬指针 #ifndef _NET_POLLER_HH #define _NET_POLLER_HH #include #include #include "TimeStamp.hh" #include "EventLoop.hh" #include "Channel.hh" struct pollfd; class Poller{ public: typedef std::vector ChannelList; Poller(EventLoop* loop); ~Poller(); TimeStamp poll(int timeoutMs, ChannelList* activeChannels); void updateChannel(Channel* channel); void assertInLoopThread() { m_pOwerLoop->assertInLoopThread(); } private: Poller& operator=(const Poller&); Poller(const Poller&); void fillActiveChannels(int numEvents, ChannelList* activeChannels) const; typedef std::vector PollFdList; typedef std::map ChannelMap; EventLoop* m_pOwerLoop; PollFdList m_pollfds; ChannelMap m_channels; }; #endif //Poller.cpp #include "Poller.hh" #include "Logger.hh" #include #include #include Poller::Poller(EventLoop* loop) : m_pOwerLoop(loop) { } Poller::~Poller() { } TimeStamp Poller::poll(int timeoutMs, ChannelList* activeChannels) { LOG_TRACE << "Poller::poll()"; int numEvents = ::poll(/*&*m_pollfds.begin()*/m_pollfds.data(), m_pollfds.size(), timeoutMs); TimeStamp now(TimeStamp::now()); if(numEvents > 0){ LOG_TRACE << numEvents << " events happended"; fillActiveChannels(numEvents, activeChannels); } else if(numEvents == 0){ LOG_TRACE << " nothing happended"; } else{ LOG_SYSERR << "Poller::poll()"; } return now; } /* *fillActiveChannels()遍历m_pollfds, 找出有活动事件的fd, 把它对应 *的Channel填入activeChannels。 */ void Poller::fillActiveChannels(int numEvents, ChannelList* activeChannels) const { for(PollFdList::const_iterator pfd = m_pollfds.begin(); pfd != m_pollfds.end() && numEvents > 0; ++pfd) { if(pfd->revents > 0) { --numEvents; ChannelMap::const_iterator ch = m_channels.find(pfd->fd); assert(ch != m_channels.end()); Channel* channel = ch->second; assert(channel->fd() == pfd->fd); channel->set_revents(pfd->revents); activeChannels->push_back(channel); } } } void Poller::updateChannel(Channel* channel) { assertInLoopThread(); LOG_TRACE << "fd= " << channel->fd() << " events" << channel->events(); if(channel->index() < 0){ //a new one , add to pollfds assert(m_channels.find(channel->fd()) == m_channels.end()); struct pollfd pfd; pfd.fd = channel->fd(); pfd.events = static_cast(channel->events()); pfd.revents = 0; m_pollfds.push_back(pfd); int idx = static_cast(m_pollfds.size()) - 1; channel->set_index(idx); m_channels[pfd.fd] = channel; } else{ //update existing one assert(m_channels.find(channel->fd()) != m_channels.end()); assert(m_channels[channel->fd()] == channel); int idx = channel->index(); assert(0 <= idx && idx < static_cast(m_pollfds.size())); struct pollfd& pfd = m_pollfds[idx]; assert(pfd.fd == channel->fd() || pfd.fd == -1); pfd.events = static_cast(channel->events()); pfd.revents = 0; if(channel->isNoneEvent()){ //ignore this pollfd pfd.fd = -1; } } } EventLoop EventLopp在上一篇文章写过,这里给出改动. EventLoop 新增了quit()成员函数, 还加了几个数据成员,并在构造函数里初始化它们。注意EventLoop通过智能指针来间接持有poller. +class Poller; +class Channel; class EventLoop ------------ bool isInloopThread() const {return m_threadId == CurrentThread::tid(); } +void quit(); +void updateChannel(Channel* channel); static EventLoop* getEventLoopOfCurrentThread(); private: EventLoop& operator=(const EventLoop&); EventLoop(const EventLoop&); void abortNotInLoopThread(); +typedef std::vector ChannelList; bool m_looping; +bool m_quit; const pid_t m_threadId; +std::unique_ptr m_poller; +ChannelList m_activeChannels; }; //EventLoop.cpp m_threadId(CurrentThread::tid()), + m_poller(new Poller(this)) { ------ +void EventLoop::quit() +{ + m_quit = true; + //wakeup(); +} + +void EventLoop::updateChannel(Channel* channel) +{ + assert(channel->ownerLoop() == this); + assertInLoopThread(); + m_poller->updateChannel(channel); +} 上一篇文章的EventLoop->loop()什么也没做,现在它有了实实在在的使命,它调用Poller::poll()获得当前活动事件的Chanel列表, 然后依次调用每个Channel的handleEvent()函数 void EventLoop::loop() { assert(!m_looping); assertInLoopThread(); m_looping = true; m_quit = false; LOG_TRACE << "EventLoop " << this << " start loopig"; while(!m_quit) { m_activeChannels.clear(); m_poller->poll(1000, &m_activeChannels); for(ChannelList::iterator it = m_activeChannels.begin(); it != m_activeChannels.end(); ++it) { (*it)->handleEvent(); } } LOG_TRACE << "EventLoop " << this << " stop loopig"; m_looping = false; } Reactor时序图 测试程序-单次触发的定时器 程序利用timerfd_create 把时间变成了一个文件描述符,该“文件”在定时器超时的那一刻变得可读,这样就能很方便地融入到 select/poll 框架中,用统一的方式来处理 IO 事件和超时事件,这也正是 Reactor 模式的长处。 #include #include #include #include "EventLoop.hh" #include "Channel.hh" #include "Poller.hh" //Reactor Test //单次触发定时器 #include EventLoop* g_loop; void timeout() { printf("timeout!\n"); g_loop->quit(); } int main() { EventLoop loop; g_loop = &loop; int timerfd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK |TFD_CLOEXEC); Channel channel(&loop, timerfd); channel.setReadCallBack(timeout); channel.eableReading(); struct itimerspec howlong; bzero(&howlong, sizeof howlong); howlong.it_value.tv_sec = 3; timerfd_settime(timerfd, 0, &howlong, NULL); loop.loop(); close(timerfd); } ./test.out 2018-10-31 22:25:54.532487 [TRACE] [EventLoop.cpp:16] [EventLoop] EventLoop Create 0x7FFEB9567CC0 in thread 3075 2018-10-31 22:25:54.533563 [TRACE] [Poller.cpp:64] [updateChannel] fd= 3 events3 2018-10-31 22:25:54.534000 [TRACE] [EventLoop.cpp:41] [loop] EventLoop 0x7FFEB9567CC0 start loopig 2018-10-31 22:25:54.534334 [TRACE] [Poller.cpp:20] [poll] Poller::poll() 2018-10-31 22:25:55.535827 [TRACE] [Poller.cpp:28] [poll] nothing happended 2018-10-31 22:25:55.536287 [TRACE] [Poller.cpp:20] [poll] Poller::poll() 2018-10-31 22:25:56.538334 [TRACE] [Poller.cpp:28] [poll] nothing happended 2018-10-31 22:25:56.538802 [TRACE] [Poller.cpp:20] [poll] Poller::poll() 2018-10-31 22:25:57.534175 [TRACE] [Poller.cpp:24] [poll] 1 events happended timeout! 2018-10-31 22:25:57.534766 [TRACE] [EventLoop.cpp:55] [loop] EventLoop 0x7FFEB9567CC0 stop loopig 作者 —— 艾露米婭娜 出处:http://www.cnblogs.com/ailumiyana/ 除特别注明外,本站所有文章均为艾露米婭娜原创,欢迎转载分享,但请注明出处。https://www.cnblogs.com/ailumiyana/p/9886358.html
50000+
5万行代码练就真实本领
17年
创办于2008年老牌培训机构
1000+
合作企业
98%
就业率

联系我们

电话咨询

0532-85025005

扫码添加微信