muduo源码分析：Reactor模式的封装实现

柔情只为你懂 2022-05-15 18:07 242阅读 0赞

关于muduo实现的Reactor模式，有三个关键的类：

1.事件分发器类Channel
    2.封装I/O复用的Poller (主要研究EpollPoller)
    3.定时器接口类TimerQueue

事件循环EventLoop类实现了Reactor的基本模式。

### 1.事件分发类Channel ###

事件分发器Channel的数据成员如下:

//定义事件类型变量
                static const int kNoneEvent;    //无事件
                static const int kReadEvent;    //可读事件
                static const int kWriteEvent;   //可写事件
    
                EventLoop *loop_;               //channel所属的loop
                const int fd_;                  //channel负责的文件描述符
                int events_;                    //注册的事件
                int revents_;                   //poller设置的就绪的事件
                int index_;                     //被poller使用的下标
                bool logHup_;                   //是否生成某些日志
    
                boost::weak_ptr<void> tie_; //
                bool tied_;
                bool eventHandling_;                //是否处于处理事件中
                bool addedToLoop_;
                ReadEventCallback readCallback_;    //读事件回调
                EventCallback writeCallback_;       //写事件回调
                EventCallback closeCallback_;       //关闭事件回调
                ReadEventCallback errorCallback_;   //错误事件回调

其中EventCallback和ReadEventCallback的声明如下:

typedef boost::function<void()> EventCallback;        //事件回调函数对象类型
     typedef boost::function<void(Timestamp)> ReadEventCallback;    //读事件回调函数对象类型

//处理事件
            void handleEvent(Timestamp receiveTime);
    
            //设置可读事件回调
            void setReadCallback(const ReadEventCallback &cb)
            {
                readCallback_ = cb;
            }
    
            //设置可写事件回调
            void setWriteCallback(const EventCallback &cb)
            {
                writeCallback_ = cb;
            }
    
            //设置关闭事件回调
            void setCloseCallback(const EventCallback &cb)
            {
                closeCallback_ = cb;
            }
    
            //设置错误事件回调
            void setErrorCallback(const EventCallback &cb)
            {
                errorCallback_ = cb;
            }
    
            void tie(const boost::shared_ptr<void>&);
    
    
    
            //返回注册的事件
            int events()const
            {
                return events_;
            }
    
            //设置就绪的事件
            void set_revents(int revt)
            {
                revents_ = revt;
            }
    
            //判断是否注册的事件
            bool isNoneEvent()const
            {
                return events_ == kNoneEvent;
            }
    
            //注册可读事件
            void enableReading()
            {
                events_ |= kReadEvent;
                update();
            }
    
            //销毁读事件
            void disableReading()
            {
                events_ &= ~kReadEvent;
                update();
            }
    
            //注册写事件
            void enableWriting()
            {
                events_ |= kWriteEvent;
             update();   
            }
    
            //销毁写事件
            void disableWriting()
            {
                events_ &= ~kWriteEvent;
                update();
            }
    
            //销毁所有事件
            void disableAll()
            {
                events_ = kNoneEvent;
                update();
            }
    
            //是否注册可写事件
            bool isWriting() const
            {
                return events_ & kWriteEvent;
            }
    
            //是否注册可读事件
            bool isReading() const
            {
                return events_ & kReadEvent;
            }

Channel的主要功能为管理各种注册给poller的套接字描述符及其上发生的事件，以及事件发生了调用事件的回调函数。

Channel的主要作用如下：

1.首先我们给定Channel所属的loop以及其要处理的fd
    2.接着我们开始注册fd_上需要监听的事件,如果是常用事件(读写等)的话，我们可以直接调用接口enable***来注册对应fd上的事件，与之对应的是disable***用来销毁特定的事件
    3.再然后我们通过set***Callback来事件发生时的回调

Channel完整代码及分析见此篇博客：

[muduo源码剖析：Channel类][muduo_Channel]

### 2.I/O复用类Poller ###

Poller类是个基类，它的定义如下:

class Poller : boost::noncopyable
        {
            public:
                typedef std::vector<Channel *> ChannelList;
    
                Poller(EventLoop *loop);
    
                virtual ~Poller();
    
                //不许在I/O线程中调用,I/O复用的封装
                virtual Timestamp poll(int timeoutMs,ChannelList *activeChannels) = 0;
    
                //更新Channel
                virtual void updateChannel(Channel *channel) = 0;
    
                //移除Channel
                virtual void removeChannel(Channel *channel) = 0;
    
                //这个channel是否在map中存在
                virtual bool hasChannel(Channel *channel)const;
    
                //默认poller方式
                static Poller *newDefaultPoller(EventLoop *loop);
    
                void assertInLoopThread()const
                {
                    ownerLoop_->assertInLoopThread();
                }
    
            protected:
                typedef std::map<int,Channel*> ChannelMap;    //map关联容器的关键字为channel_所管理的fd
                ChannelMap Channels_;                   //存储事件分发器的map
            private:
                EventLoop *owerLoop_;                   //属于哪个loop
        };

事件分发器channel集用关联容器map来保存，map的关键字为channel所管理的fd。

看下epoll对Poller的实现

EpollPoller类的定义为: EpollPoller.h

class EPollPoller : public Poller
    {
    public:
        EPollPoller(EventLoop* loop);
        virtual ~EPollPoller();
    
        virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels);	//内部调用epoll_wait函数
        virtual void updateChannel(Channel* channel);    //添加更新事件			
        virtual void removeChannel(Channel* channel);    //移除事件
    
    private:
        static const int kInitEventListSize = 16;			//epoll事件表的大小
    
        static const char* operationToString(int op);
    
        void fillActiveChannels(int numEvents,
                              ChannelList* activeChannels) const;	//将epoll返回的活跃事件填充到activeChannel
        void update(int operation, Channel* channel);		//注册事件
    
        typedef std::vector<struct epoll_event> EventList;	//typedef:事件数组
    
        int epollfd_;			//epoll的事件表fd: epoll例程
        EventList events_;	    //epoll事件数组
    };

Poller主要功能如下：

1.调用poll函数监听注册了事件的文件描述符
    2.当poll返回时将发生事件的事件集装入activeChannels中,并设置Channel发生事件到其revents_中
    3.控制channel中事件的增删改

EPollPoller完整实现以及分析见此篇博客：

[muduo源码分析：Poller类 (EpollPoller类)][muduo_Poller_ _EpollPoller]

### 3.定时器TimerQueue ###

EventLoop直接调用的是定时器队列类TimerQueue类，该类的数据定义如下:

typedef std::pair<Timestamp,Timer *> Entry;     //定时器入口
                typedef sta::set<Entry> TimerList;              //定时器集合
                typedef std::pair<Timer *,int64_t> ActiveTimer; //到时的定时器
                typedef std::set<ActiveTimeri> ActiveTimerSet;  //活跃的定时器集合
    
    EventLoop *loop_;       //定时器所在的I/O循环
                const int timerfd_;     //定时器描述符
                Channel timerfdChannel_;
    
                TimerList timers_;      //定时器集合
    
                //for cancel()
                ActiveTimerSet activeTimers_;
                bool callingExpiredTimers_;
                ActiveTimerSet cancelingTimers_;

用来存储定时器集的容器使用了set，set针对定时器有它天然的优势，首先set的特性是所存储的元素为默认升序的，这样当我们某个事件点取到期的定时器，就直接取该时间点之前的所有定时器就好，其次我们往set中添加定时器的效率也相对较好为0(logn)。但是用set有个问题，我们如何存储俩个定时时间一样的定时器呢？muduo的解决方案就是使用一个pair类型。

//增加一个定时器
     TimerId addTimer(const TimerCallback &cb,
                                Timestamp when,
                                double interval);
     //删除一个定时器
     void cancel(TimerId timerId);

## 核心： ##

### 4.EventLoop类的实现 ###

EventLoop类实现了reactor的基本模式 ，它的数据定义如下：

void abortNotInLoopThread();				//不在主I/O线程 
      void handleRead();  // waked up			//将事件通知描述符里的内容读走,以便让其继续检测事件通知
      void doPendingFunctors();			        //执行转交给I/O的任务 
    
      void printActiveChannels() const; // DEBUG	        //将发生的事件写入日志
    
      typedef std::vector<Channel*> ChannelList;		//事件分发器列表
    
      bool looping_; /* atomic */				//是否运行
      bool quit_; /* atomic and shared between threads, okay on x86, I guess. */	//是否退出事件循环
      bool eventHandling_; /* atomic */
      bool callingPendingFunctors_; /* atomic */
      int64_t iteration_;					//事件循环的次数
      const pid_t threadId_;				//运行loop的线程ID
      Timestamp pollReturnTime_;				//poll阻塞的时间
      boost::scoped_ptr<Poller> poller_;			//IO复用
      boost::scoped_ptr<TimerQueue> timerQueue_;		//定时器队列
      int wakeupFd_;					//唤醒套接字
      // unlike in TimerQueue, which is an internal class,
      // we don't expose Channel to client.
      boost::scoped_ptr<Channel> wakeupChannel_;		//封装事件描述符
      boost::any context_;
    
      // scratch variables
      ChannelList activeChannels_;				//活跃的事件集
      Channel* currentActiveChannel_;			//当前处理的事件集
    
      mutable MutexLock mutex_;				//互斥锁
      std::vector<Functor> pendingFunctors_ GUARDED_BY(mutex_);	//需要在主I/O线程执行的任务

EventLoop通过boost库下的智能指针scoped\_ptr来管理Poller\_,TimerQueue\_，wakeupChannel\_对象，这样不容易发生内存显露，其中变量pendingFunctors\_为需要在I/O线程中执行的任务集，例如上面所讲的定时器的增删接口的执行，就会先放在此集合里，然后有主I/O线程来执行，那么主线程在调用loop函数之后会阻塞在poller函数中，此时我们应该如何唤醒I/O线程呢？muduo中采用了linux的新特性eventfd来唤醒I/O线程。

EventLoop的主要功能如下：

1.首先我们应该调用updateChannel来添加一些事件(内部调用poller->updateChannel()来添加注册事件)
    2.接着调用loop函数来执行事件循环，在执行事件循环的过程中，会则色在poller->poll调用处，Poller类会把活跃的事件放在activeChannel集合中
    3.然后调用Channel中的handleEvent来处理事件发生时对应的回调函数，处理完事件函数后还会处理必须由I/O线程来完成的doPendingFunctors函数

当然我们可以在中间的过程中注册一些普通事件或通过run\*类函数来注册定时事件，我们也可以调用updateChannel和removeChannel来增删Channel。

EventLoop的完整源码如下：

**EventLoop.h:**

#ifndef MUDUO_NET_EVENTLOOP_H
    #define MUDUO_NET_EVENTLOOP_H
    
    #include <vector>
    
    #include <boost/any.hpp>
    #include <boost/function.hpp>
    #include <boost/noncopyable.hpp>
    #include <boost/scoped_ptr.hpp>
    
    #include <muduo/base/Mutex.h>
    #include <muduo/base/CurrentThread.h>
    #include <muduo/base/Timestamp.h>
    #include <muduo/net/Callbacks.h>
    #include <muduo/net/TimerId.h>
    
    namespace muduo
    {
    namespace net
    {
    
    class Channel;
    class Poller;
    class TimerQueue;
    
    /// Reactor, at most one per thread.
    ///
    /// This is an interface class, so don't expose too much details.
    
    
    class EventLoop : boost::noncopyable
    {
     public:
      typedef boost::function<void()> Functor;
    
      EventLoop();
      ~EventLoop();  // force out-line dtor, for scoped_ptr members.
    
    
      /// 主循环
      void loop();
    
      /// 退出主循环
      void quit();
    
      /// poll延迟的时间
      Timestamp pollReturnTime() const { return pollReturnTime_; }
    
      /// 迭代次数
      int64_t iteration() const { return iteration_; }
     
      /// 在主循环中运行
      void runInLoop(const Functor& cb);
     
      /// 插入主循环任务队列
      void queueInLoop(const Functor& cb);
    
      size_t queueSize() const;
    
      /// 某个时间点执行定时回调
      TimerId runAt(const Timestamp& time, const TimerCallback& cb);
    
      /// 某个时间点之后执行定时回调
      TimerId runAfter(double delay, const TimerCallback& cb);
     
      /// 在每个时间间隔处理某个回调函数
      TimerId runEvery(double interval, const TimerCallback& cb);
     
      /// 删除某个定时器
      void cancel(TimerId timerId);
    
      void wakeup();		                //唤醒事件通知描述符
      void updateChannel(Channel* channel);		//添加某个事件分发器
      void removeChannel(Channel* channel);		//移除某个事件分发器
      bool hasChannel(Channel* channel);
    
      // pid_t threadId() const { return threadId_; }
      void assertInLoopThread()			//如果不在I/O线程中则退出程序
      {
        if (!isInLoopThread())
        {
          abortNotInLoopThread();
        }
      }
      bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }	//检测是否在I/O线程中
      // bool callingPendingFunctors() const { return callingPendingFunctors_; }
      bool eventHandling() const { return eventHandling_; }				//是否正在处理事件
    
      void setContext(const boost::any& context)
      { context_ = context; }
    
      const boost::any& getContext() const
      { return context_; }
    
      boost::any* getMutableContext()
      { return &context_; }
    
      static EventLoop* getEventLoopOfCurrentThread();	//判断当前线程是否为I/O线程
    
     private:
      void abortNotInLoopThread();				//不在主I/O线程 
      void handleRead();  // waked up			//将事件通知描述符里的内容读走,以便让其继续检测事件通知
      void doPendingFunctors();				//执行转交给I/O的任务 
    
      void printActiveChannels() const; // DEBUG	        //将发生的事件写入日志
    
      typedef std::vector<Channel*> ChannelList;	        //事件分发器列表
    
      bool looping_; /* atomic */				//是否运行
      bool quit_; 	                        //是否退出事件循环
      bool eventHandling_; /* atomic */
      bool callingPendingFunctors_; /* atomic */
      int64_t iteration_;					//事件循环的次数
      const pid_t threadId_;				//运行loop的线程ID
      Timestamp pollReturnTime_;			        //poll阻塞的时间
      boost::scoped_ptr<Poller> poller_;		        //IO复用
      boost::scoped_ptr<TimerQueue> timerQueue_;		//定时器队列
      int wakeupFd_;					//唤醒套接字
      // unlike in TimerQueue, which is an internal class,
      // we don't expose Channel to client.
      boost::scoped_ptr<Channel> wakeupChannel_;		//封装事件描述符
      boost::any context_;
    
      // scratch variables
      ChannelList activeChannels_;				//活跃的事件集
      Channel* currentActiveChannel_;			//当前处理的事件集
    
      mutable MutexLock mutex_;				//互斥锁
      std::vector<Functor> pendingFunctors_ GUARDED_BY(mutex_);	//需要在主I/O线程执行的任务
    };
    
    }
    }
    #endif  // MUDUO_NET_EVENTLOOP_H

**EventLoop.cc**

#include <muduo/net/EventLoop.h>
    
    #include <muduo/base/Logging.h>
    #include <muduo/base/Mutex.h>
    #include <muduo/net/Channel.h>
    #include <muduo/net/Poller.h>
    #include <muduo/net/SocketsOps.h>
    #include <muduo/net/TimerQueue.h>
    
    #include <boost/bind.hpp>
    
    #include <signal.h>
    #include <sys/eventfd.h>
    #include <unistd.h>
    
    using namespace muduo;
    using namespace muduo::net;
    
    namespace
    {
    	//当前线程EventLoop对象指针
    	//线程局部存储
    __thread EventLoop* t_loopInThisThread = 0;
    
    const int kPollTimeMs = 10000;
    
    int createEventfd()
    {
      int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
      if (evtfd < 0)
      {
        LOG_SYSERR << "Failed in eventfd";
        abort();
      }
      return evtfd;
    }
    
    #pragma GCC diagnostic ignored "-Wold-style-cast"
    class IgnoreSigPipe
    {
     public:
      IgnoreSigPipe()
      {
        ::signal(SIGPIPE, SIG_IGN);
        // LOG_TRACE << "Ignore SIGPIPE";
      }
    };
    #pragma GCC diagnostic error "-Wold-style-cast"
    
    IgnoreSigPipe initObj;
    }
    
    EventLoop* EventLoop::getEventLoopOfCurrentThread()
    {
      return t_loopInThisThread;
    }
    
    EventLoop::EventLoop()		//初始化事件循环
      : looping_(false),
        quit_(false),
        eventHandling_(false),
        callingPendingFunctors_(false),
        iteration_(0),
        threadId_(CurrentThread::tid()),
        poller_(Poller::newDefaultPoller(this)),
        timerQueue_(new TimerQueue(this)),
        wakeupFd_(createEventfd()),
        wakeupChannel_(new Channel(this, wakeupFd_)),
        currentActiveChannel_(NULL)
    {
      LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;
      if (t_loopInThisThread)
      {
        LOG_FATAL << "Another EventLoop " << t_loopInThisThread
                  << " exists in this thread " << threadId_;
      }
      else
      {
        t_loopInThisThread = this;
      }
      wakeupChannel_->setReadCallback(
          boost::bind(&EventLoop::handleRead, this));		//注册读完成时的回调函数
      // we are always reading the wakeupfd
      wakeupChannel_->enableReading();				//注册可读事件
    }
    
    EventLoop::~EventLoop()
    {
      LOG_DEBUG << "EventLoop " << this << " of thread " << threadId_
                << " destructs in thread " << CurrentThread::tid();
      wakeupChannel_->disableAll();
      wakeupChannel_->remove();
      ::close(wakeupFd_);
      t_loopInThisThread = NULL;
    }
    
    void EventLoop::loop()
    {
      assert(!looping_);
      assertInLoopThread();
      looping_ = true;
      quit_ = false;  // FIXME: what if someone calls quit() before loop() ?
      LOG_TRACE << "EventLoop " << this << " start looping";
    
      while (!quit_)
      {
        activeChannels_.clear();	//删除事件列表所有元素
    //通过poller获取就绪的channel，放到activeChannels_中,poller会将发生的事件类型填写到channel的revents_中，供Channel::handleEvent使用,调用相应的事件回调函数
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
        ++iteration_;
        if (Logger::logLevel() <= Logger::TRACE)
        {
          printActiveChannels();	//将发生的事件写入日志
        }
        // TODO sort channel by priority
        eventHandling_ = true;
        
        //处理就绪事件
        for (ChannelList::iterator it = activeChannels_.begin();
            it != activeChannels_.end(); ++it)
        {
          currentActiveChannel_ = *it;
    	  //调用channel的事件处理函数handleEvent,根据poller设置的发生的事件类型，调用相应的用户函数
          currentActiveChannel_->handleEvent(pollReturnTime_);
        }
        currentActiveChannel_ = NULL;
        eventHandling_ = false;
        doPendingFunctors();		//处理一些其它的任务
      }
    
      LOG_TRACE << "EventLoop " << this << " stop looping";
      looping_ = false;
    }
    
    void EventLoop::quit()
    {
      quit_ = true;
      // There is a chance that loop() just executes while(!quit_) and exits,
      // then EventLoop destructs, then we are accessing an invalid object.
      // Can be fixed using mutex_ in both places.
      if (!isInLoopThread())
      {
        wakeup();
      }
    }
    
    void EventLoop::runInLoop(const Functor& cb)
    {
      if (isInLoopThread())
      {
        cb();
      }
      else
      {
        queueInLoop(cb);
      }
    }
    
    void EventLoop::queueInLoop(const Functor& cb)
    {
      {
      MutexLockGuard lock(mutex_);
      pendingFunctors_.push_back(cb);
      }
    
      if (!isInLoopThread() || callingPendingFunctors_)
      {
        wakeup();
      }
    }
    
    size_t EventLoop::queueSize() const
    {
      MutexLockGuard lock(mutex_);
      return pendingFunctors_.size();
    }
    
    TimerId EventLoop::runAt(const Timestamp& time, const TimerCallback& cb)
    {
      return timerQueue_->addTimer(cb, time, 0.0);
    }
    
    TimerId EventLoop::runAfter(double delay, const TimerCallback& cb)
    {
      Timestamp time(addTime(Timestamp::now(), delay));
      return runAt(time, cb);
    }
    
    TimerId EventLoop::runEvery(double interval, const TimerCallback& cb)
    {
      Timestamp time(addTime(Timestamp::now(), interval));
      return timerQueue_->addTimer(cb, time, interval);
    }
    
    void EventLoop::cancel(TimerId timerId)
    {
      return timerQueue_->cancel(timerId);
    }
    
    void EventLoop::updateChannel(Channel* channel)		//添加更新事件分发器到map中
    {
      assert(channel->ownerLoop() == this);
      assertInLoopThread();
      poller_->updateChannel(channel);
    }
    
    void EventLoop::removeChannel(Channel* channel)		//从map中移除事件分发器
    {
      assert(channel->ownerLoop() == this);
      assertInLoopThread();
      if (eventHandling_)
      {
        assert(currentActiveChannel_ == channel ||
            std::find(activeChannels_.begin(), activeChannels_.end(), channel) == activeChannels_.end());
      }
      poller_->removeChannel(channel);
    }
    
    bool EventLoop::hasChannel(Channel* channel)		//查找事件分发器是否在channels_中
    {
      assert(channel->ownerLoop() == this);
      assertInLoopThread();
      return poller_->hasChannel(channel);
    }
    
    void EventLoop::abortNotInLoopThread()
    {
      LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
                << " was created in threadId_ = " << threadId_
                << ", current thread id = " <<  CurrentThread::tid();
    }
    
    void EventLoop::wakeup()
    {
      uint64_t one = 1;
      ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);
      if (n != sizeof one)
      {
        LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
      }
    }
    
    void EventLoop::handleRead()
    {
      uint64_t one = 1;
      ssize_t n = sockets::read(wakeupFd_, &one, sizeof one);
      if (n != sizeof one)
      {
        LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
      }
    }
    
    void EventLoop::doPendingFunctors()
    {
      std::vector<Functor> functors;
      callingPendingFunctors_ = true;
    
      {
      MutexLockGuard lock(mutex_);
      functors.swap(pendingFunctors_);        //提高效率防止死锁
      }
    
      for (size_t i = 0; i < functors.size(); ++i)
      {
        functors[i]();
      }
      callingPendingFunctors_ = false;
    }
    
    void EventLoop::printActiveChannels() const		//将发生的事件写入日志
    {
      for (ChannelList::const_iterator it = activeChannels_.begin();
          it != activeChannels_.end(); ++it)
      {
        const Channel* ch = *it;
        LOG_TRACE << "{" << ch->reventsToString() << "} ";
      }
    }

[muduo_Channel]: https://blog.csdn.net/amoscykl/article/details/83552506
[muduo_Poller_ _EpollPoller]: https://blog.csdn.net/amoscykl/article/details/83552535