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

电玩女神 2022-04-22 00:12 119阅读 0赞

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

### 文章目录 ###

*  muduo学习笔记(二)Reactor关键结构
 *   *  Reactor简述
     *   *  什么是Reactor
         *  Reactor模型的优缺点
     *  poll简述
     *   *  poll使用样例
     *  muduo Reactor关键结构
     *   *  Channel
         *  Poller
         *  EventLoop
         *  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 <poll.h>
    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 <fcntl.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <stdlib.h>
    #include <string.h>
    #include <time.h>
    #include <errno.h>
    #include <poll.h>
    
    #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）的事件处理函数。

![Reactor关键结构UML图][Reactor_UML]

### 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 <functional>
    
    #include "EventLoop.hh"
    
    class Channel {
        
    public:
      typedef std::function<void()> 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 <poll.h>
    #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 <vector>
    #include <map>
    
    #include "TimeStamp.hh"
    #include "EventLoop.hh"
    #include "Channel.hh"
    
    struct pollfd;
    
    class Poller{
        
    public:
      typedef std::vector<Channel*> 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<struct pollfd> PollFdList;
      typedef std::map<int, Channel*> ChannelMap;
    
      EventLoop* m_pOwerLoop;
      PollFdList m_pollfds;
      ChannelMap m_channels;
    
    };
    
    #endif
    
    //Poller.cpp
    #include "Poller.hh"
    #include "Logger.hh"
    #include <assert.h>
    #include <poll.h>
    #include <signal.h>
    
    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<short>(channel->events());
        pfd.revents = 0;
        m_pollfds.push_back(pfd);
        int idx = static_cast<int>(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<int>(m_pollfds.size()));
        struct pollfd& pfd = m_pollfds[idx];
        assert(pfd.fd == channel->fd() || pfd.fd == -1);
        pfd.events = static_cast<short>(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<Channel*> ChannelList;
    
    	bool m_looping;
    	+bool m_quit;
    	const pid_t m_threadId;
    	+std::unique_ptr<Poller> 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时序图 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MzA4MzIx_size_16_color_FFFFFF_t_70]

## 测试程序-单次触发的定时器 ##

程序利用timerfd\_create 把时间变成了一个文件描述符，该“文件”在定时器超时的那一刻变得可读，这样就能很方便地融入到 select/poll 框架中，用统一的方式来处理 IO 事件和超时事件，这也正是 Reactor 模式的长处。

#include <errno.h>
    #include <thread>
    #include <strings.h>
    
    #include "EventLoop.hh"
    #include "Channel.hh"
    #include "Poller.hh"
    
    //Reactor Test
    //单次触发定时器
    #include <sys/timerfd.h>
    
    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

[Reactor_UML]: /images/20220422/1103bb04b69a42718bb3fa659b39fff8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE3MzA4MzIx_size_16_color_FFFFFF_t_70]: /images/20220422/1c55f44d3c0546febb04bc69e8773d41.png