muduo源码分析之定时器TimerQueue的设计与实现

逃离我推掉我的手 2022-07-14 04:18 287阅读 0赞

## **1.简介** ##

这部分介绍`TimerId`、`Timer`、`TimerQueue`三个class的封装，反映到实际使用，主要是`EventLoop`中的三个函数：`runAt()`、`runAfter()`、`runEvery()`。

`TimerQueue`的封装是为了让未到期的时间`Timer`有序的排列起来，这样，能够更具当前时间找到已经到期的`Timer`也能高效的添加和删除`Timer`。

所谓的到期与未到期，与当前在当前时间之前表示已经到期，之后则是未到期。为了方便计算，muduo重载了`operator<`主要是用来比较微秒大小。

到期的时间应该被清除去执行相应的回调，未到期的时间则应该有序的排列起来。

`TimerQueue`的封装是为了让未到期的时间`Timer`有序的排列起来，这样，能够更具当前时间找到已经到期的`Timer`也能高效的添加和删除`Timer`。

对于`TimerQueue`的数据结构，作者提出了几个方案。  
1.传统线性表，查找复杂度为O(n)  
2.二叉堆实现优先级队列，不过C++标准的`make_heap()`不能高效地完成删除操作。

最终，为了防止时间相同所导致的Key相同的情况，使用set和pair

typedef std::pair<Timestamp, Timer*>Entry;
    typedef std::set<Entry>TimerList;
    TimerList timers_;

这样既可以通过一个时间点和超时事件一一对应，又可以处理多个时间相同的情况。

## **2.timerfd\_\* 和gettimeofday** ##

这两个函数是muduo进行封装的底层实现。

### 2.1使用timerfd\_系列函数来处理定时任务 ###

主要有三个接口：

#include <sys/timerfd.h>
    int timerfd_create(int clockid, int flags);
    int timerfd_settime(int fd, int flags,
                               const struct itimerspec *new_value,
                               struct itimerspec *old_value);
    int timerfd_gettime(int fd, struct itimerspec *curr_value);

通过`timerfd_settime`下面的结构体，可以设置超时时间，和超时的重复时间（每隔多少时间超时）。

struct timespec {
        time_t tv_sec;                /* Seconds */
        long   tv_nsec;               /* Nanoseconds */
               };
    
    struct itimerspec {
        struct timespec it_interval;  /* Interval for periodic timer */
        struct timespec it_value;     /* Initial expiration */
               };

当有超时事件发生的时候，由`timerfd_create`返回的描述符就可读，我们可以通过`read(2)`获取当前的超时事件的数量，由一个8个字节的变量返回。  
***read(2) If the timer has already expired one or more times since its settings were last modified using timerfd\_settime(), or since the last successful read(2), then the buffer given to read(2) returns an unsigned 8-byte integer (uint64\_t) containing the number of expirations that have occurred.***

### 2.2使用gettimeofday来获取当前时间 ###

这个函数被封装在`Timestamp`类中，简单示意如下：

class Timestamp
    {
        public:
            Timestamp():microSecondsSinceEpoch_(0){}
            Timestamp(uint_64 micro):microSecondsSinceEpoch_(mirco){}
            static Timestamp now(){
                struct timeval tv; 
                gettimeofday(&tv, NULL);
                int64_t seconds = tv.tv_sec;
                //该构造初始化了microSecsSinceEpoch_
                return Timestamp(seconds * kMicroSecondsPerSecond + tv.tv_usec);
            }
    };

通过把时间进行微秒级别的量化从而方便对时间戳的比较。

## **3.Timer\* 类的设计与实现** ##

这里涉及了3个类`TimerId`、`Timer`、`TimerQueue`要知道他们怎么与`EventLoop`结合起来，还是从一个示例开始。简单来说，TimerQueue是用来进行管理调度的，而Timer是真正的超时事件（该Class中封装了真正的超时回调）

void printTid()
    {
      printf("pid = %d, tid = %d\n", getpid(), muduo::CurrentThread::tid());
      printf("now %s\n", muduo::Timestamp::now().toString().c_str());
    }
    
    void print(const char* msg)
    {
      printf("msg %s %s\n", muduo::Timestamp::now().toString().c_str(), msg);
      if (++cnt == 20)
      {
        g_loop->quit();
      }
    }
    
    int main()
    {
      printTid();
      muduo::EventLoop loop;
      g_loop = &loop;
    
      print("main");
      loop.runAfter(1, boost::bind(print, "once1"));
    #if 1
      loop.runAfter(1.5, boost::bind(print, "once1.5"));
      loop.runAfter(2.5, boost::bind(print, "once2.5"));
      loop.runAfter(3.5, boost::bind(print, "once3.5"));
      loop.runEvery(2, boost::bind(print, "every2"));
      loop.runEvery(3, boost::bind(print, "every3"));
    #endif
      loop.loop();
      print("main loop exits");
      sleep(1);
    }

首先，是`EventLoop`类对象的初始化：

EventLoop::EventLoop()
      : looping_(false),
        quit_(false),
        threadId_(CurrentThread::tid()),
        poller_(new Poller(this)),
        timerQueue_(new TimerQueue(this))
    {
    //...
    }

这里，我们看到在`EventLoop`的构造函数中对`Poller`和`TimerQueue`类型的成员进行初始化。

然后，看`TimerQueue`的构造函数：

TimerQueue::TimerQueue(EventLoop* loop)
      : loop_(loop),
        timerfd_(createTimerfd()),//timefd描述符
        timerfdChannel_(loop, timerfd_),//channel对象
        timers_()//保存Timer的关键结构
    {
      timerfdChannel_.setReadCallback(
          boost::bind(&TimerQueue::handleRead, this));
      // we are always reading the timerfd, we disarm it with timerfd_settime.
      timerfdChannel_.enableReading();
    }

将timefd和Channel关联起来，并绑定`TimerQueue::handleRead`作为描述符可读时的回调函数，而在`TimerQueue::handleRead`中又会去调用超时的`Timer`的回调。

void TimerQueue::handleRead()
    {
      loop_->assertInLoopThread();
      Timestamp now(Timestamp::now());
      readTimerfd(timerfd_, now);
    
      std::vector<Entry> expired = getExpired(now);//获取超时的事件
      //超时定义为比当前时间点早的时间。
    
    //再依次调用Timer中的回调。
      // safe to callback outside critical section
      for (std::vector<Entry>::iterator it = expired.begin();
          it != expired.end(); ++it)
      {
        it->second->run();
      }
      reset(expired, now);
    }

构造函数帮我们做了很多事情，不过上述情况是假设`TimerQueue`中已经有一系列已经排好序列的时间事件了，现在要来看看怎么添加这些时间。

对应`EventLoop`中的函数就是`runAt、runAfter、runEvery`  
其实主要是`TimeQueue::addTimer`函数，因为`runAfter`和`runEvery`都是通过设置不同的参数去调用`TimeQueue::addTimer`。

TimerId TimerQueue::addTimer(const TimerCallback& cb,//超时回调
                                 Timestamp when,//超时时间
                                 double interval)//重复时间
    {
      Timer* timer = new Timer(cb, when, interval);//构造Timer
      loop_->assertInLoopThread();
      bool earliestChanged = insert(timer);//插入Timer到TimerList中和Timestamp关联起来
    
      if (earliestChanged)
      {
        resetTimerfd(timerfd_, timer->expiration());
      }
      return TimerId(timer);
    }

这样看下来就清晰了：  
整个过程如下：  
1.构造函数创建timerfd，设置超时回调该回调将用来构造`Timer`  
2.通过`runAt`、`runAfter`、`runEvery`添加`Timer`和设置`Timer`的回调  
3.此时定时器已经启动，当发生超时事件时，timerfd可读，poll调用返回。  
4.通过activeChannl返回活跃事件（这里的活跃事件即为timerfd可读事件）  
5.在`Channl::handleRead()`回调中将执行在TimerQueue构造函数中绑定的`TimerQueue::handleRead()`。  
6.`TimerQueue::handleRead()`又将获取所有超时的`Timer`并执行`Timer`中的回调。

Created with Raphaël 2.1.0开始1.构造函数创建timerfd，设置超时回调该回调将用来构造Timer2.通过runAt、runAfter、runEvery添加Timer和设置Timer的回调3.loop.loop()，此时定时器已经启动，当发生超时事件时，timerfd可读，poll调用返回。4.通过activeChannl返回活跃事件（这里的活跃事件即为timerfd可读事件）5.在Channl::handleRead()回调中将执行在TimerQueue构造函数中绑定的TimerQueue::handleRead()。6.TimerQueue::handleRead()又将获取所有超时的Timer并依次执行Timer中的回调。7.执行完一次超时回调后，这个时候需要重新设置定时器，把当前未过期的最早时间作为定时器最新时间(Timestamp::reset())。结束