深入Linux C/C++ Timer定时器的实现核心原理

我曾以为像定时器这样基础的功能，操作系统会有一个完备的实现。当需要开启一个定时任务的时候，会有一个优雅的、如下形式的接口：

typedef void (*callback)(void*);
void setTimeout(unsigned int second,callback cb，void* arg);

可是事与愿违，Linux下不存在这样的接口。

文章相关视频讲解：

C/C++ Linux服务器开发高级架构学习视频点击： C/C++Linux服务器开发/Linux后台架构师-学习视频

如何实现分布式定时器及核心原理

详解Linux网络编程相关的细节处理

定时器的实现原理

定时器的实现依赖的是CPU时钟中断，时钟中断的精度就决定定时器精度的极限。一个时钟中断源如何实现多个定时器呢？对于内核，简单来说就是用特定的数据结构管理众多的定时器，在时钟中断处理中判断哪些定时器超时，然后执行超时处理动作。而用户空间程序不直接感知CPU时钟中断，通过感知内核的信号、IO事件、调度，间接依赖时钟中断。用软件来实现动态定时器常用数据结构有：时间轮、最小堆和红黑树。下面就是一些知名的实现：

Hierarchy 时间轮算法：Linux内核
红黑树最小堆算法：Asio C++ Library或nginx

Linux上的定时函数

要想使用上面那样的定时器功能，我们必须利用Linux上现有的定时通知函数，封装一个定时器。Linux上的定时通知函数五花八门，要封装我们自己的定时器，首先需要选用一个定时通知的函数。查阅资料整理出了Linux上所有的定时函数，如下表：

前四个函数比较鸡肋，会让调用线程挂起，原地等待定时器超时，否定。

alarm()和setitimer()，它们的通知机制采用了信号SIGALRM，由于SIGALRM信号不可靠，会造成超时通知不可靠，而且多线程中处理信号也是一个麻烦事，也不考虑。

timer_create()/timer_settime()系列函数是POSIX规定，精度达到纳秒级，提供了一个数据结构struct sigevent可以指定一个实时信号作为通知信号，同时也可以设置线程ID，将信号传递到指定的线程。相比前两个函数，有了不小的改进，可以作为一个备选的实现，但是可以预见到封装起来不会很轻松。此外使用此系列的函数，需要链接librt库。

事实上，我们遗漏掉了几个同样具有定时的功能的API——多路复用。在Linux上的多路复用机制有select/poll/epoll几种，它们轮询时都允许指定一个超时时间，如果在指定时间内，监控的事件没有到达，轮询函数会超时返回。精度也足够用，poll/epoll是毫秒级的(millisecond),select超时参数是struct timeval，是微秒级的(microsecond)。

选择epoll的优势很明显，能将定时功能完美的融入已有的event loop里，同时epoll有着天然的高并发的能力，millisecond级的精度也足够用。

获取当前时间

要实现一个定时器，有了定时函数，我们还需要选用一个获取时间的函数。同样地，这些函数我也整理了一下：

time()精度太低，不合适。

ftime() 毫秒级精度，但是被废弃了，也不合适。

gettimeofday() 精度达到微秒级，并且在x86-64平台上该函数的调用不是系统调用(vdso)，似乎很合适，不幸的是POSIX.1-2008中也将这个函数废弃了。

Time Stamp Counter 使用汇编指定获取时间戳的计数器，精度应该是最高的，效率 可能 也应该是最高的，一条汇编指令rdtscp(相比rdtsc，rdtscp可以避免，因为cpu乱序执行带来的误差问题)即可。是可以作为一个选择的，腾讯的libco就是优先使用这个方法获取时间的。

clock_gettime() 。默认是nanosecond 级精度，是系统调用(_sys_clock_gettime())，会有开销。调用频繁的话，可能造成损失性能。但是Linux 2.6.32后可以指定参数CLOCK_REALTIME_COARSE和CLOCK_MONOTONIC_COARSE，粗粒度地获取时间，而不需要发生上下文切换(和gettimeofday()一样也是vdso技术， https:// access.redhat.com/docum entation/en-us/red_hat_enterprise_linux_for_real_time/7/html/reference_guide/sect-posix_clocks#CLOCK_MONOTONIC_COARSE_and_CLOCK_REALTIME_COARSE )。使用_COARSE后缀获取的时间，精度是millisecond级。对我们来说也够用了。

要对clock_gettime系统调用的开销有一个直观的感受的话可以，借助于strace工具，利用-T参数可以追踪每个系统调用的时间开销，比我我的环境上CLOCK_MONOTONIC获取时间的开销大概是50微秒

换成CLOCK_MONOTONIC_COARSE方式再去获取时间，用strace就追踪不到了。

定时器的设计

有了获取时间函数clock_gettime和定时函数epoll之后，我们就可以开始设计定时器了。首先明确一点，epoll和其他的定时通知函数一样，一次也只能设置一个超时时间，依然不能满足我开篇提出的需求。

主流的做法是利用一个容器保存所有设置的超时时间，将容器里最快的超时的时间设置为epoll_wait的超时时间。比如，我先后设置了1400ms,800ms,300ms,2900ms，那么下一次事件循环就将epoll_wait的第四个参数设置为300ms。

如果用链表保存的话，每次设置定时器都要遍历一遍链表才能选到最快超时的那个时间，复杂度太高，如果设置了定时器特别多的话，这样的开销不能接受。

要像O(1)的时间获取到最小的哪个值，用最小堆保存超时时间正合适，效率大大提高。事实上libevent就是这么实现的(C语言实现的min_heap_t)。

关于C/C++ Linux后端开发网络底层原理知识 点击 学习资料 获取，内容知识点包括Linux，Nginx，ZeroMQ，MySQL，Redis，线程池，MongoDB，ZK，Linux内核，CDN，P2P，epoll，Docker，TCP/IP，协程，DPDK等等。

最小堆实现

先实现一个类Timer表示每一个被添加的定时，构造时需要一个millisecond为单位的超时时间，一个回调函数，一个回调函数的参数。为了简化实现，我测试用的超时的回调函数，并未使用回调函数的参数，但也没有去掉，仅仅是占个坑的作用。本来是想打算把args抽象，将Timer写成模板类，防止本末倒置，本文仅为演示定时器的实现，越简单越好。

expire时间用的是相对系统启动的时间,是一个不可以设置的恒定的时间(nonsettable monotonic clock)，而不是用的真实的时间(Wall time ，墙上时间)，因为这个时间可能会随着设置系统的日期时间而发生跳跃性的改变。

class Timer
public:
    Timer(unsigned long long expire, std::function<void(void)> fun, void *args)
        : expire_(expire), fun(fun){    }
    inline void active() { fun(); }
    inline unsigned long long getExpire() const{ return expire_; }
private:
    std::function<void(void)> fun;
    void *args;
    unsigned long long expire_;

TimerManager是用户操作的接口，提供增加，删除定时器的功能。STL中提供能优先队列，直接可以拿来用。

class TimerManager
public:
    TimerManager() {}
    Timer *addTimer(int timeout, std::function<void(void)> fun, void *args = NULL);
    void delTimer(Timer* timer);
    unsigned long long getRecentTimeout();
    void takeAllTimeout();
    unsigned long long getCurrentMillisecs();
private:
    struct cmp
        bool operator()(Timer*& lhs, Timer*& rhs) const { return lhs->getExpire() > rhs->getExpire(); }
    std::priority_queue<Timer*,std::vector<Timer*>,cmp> queue_;

add
Timer()参数和Timer构造函数一直，实现就是构造一个Timer然后加入到std::priority_queue后，返回Timer指针。

delTimer() 删除一个指定的Timer，由于priority_queue没有提供erease()接口，因此删除Timer的操作，我这里采用了新建一个priority_queue的做法，复杂度O(n)。

getRecentTimeout()获取一个最近的超时时间(超时时间 = 优先队列里的时间 - 当前获取的系统启动时间)。如果这个值小于0，那么说明这个定时器已经超时了，将其置为0，稍后的epoll_wait将会立马返回。

takeAllTimeout() 函数，处理超时的定时，并回调其绑定的回调函数。由于超时的可能不止一个定时，需要用一个循环遍历所有超时的Timer，一一处理。

getCurrentMillisecs()对clock_gettime()的封装，获取到的struct timespec转换为millisecond。

这两个类的完整实现，我放到了Github上了: https:// gist.github.com/baixian gcpp/b2199f1f1c7108f22f47d2ca617f6960 。使用的时候，只需要在你的主循环里，把epoll_wait的超时参数设置为TimerManager::getRecentTimeout()，每次epoll_wait()返回后，处理一下超时事件TimerManager::takeAllTimeout()。

使用示例：

int dispatch()
    TimerManager tm;
    tm.addTimer(1000, []() { std::cout << "hello world" << std::endl; }, NULL);
    tm.addTimer(5000, []() { std::cout << "hello baixiancpp" << std::endl; }, NULL);
    for(;;)
        int ret = epoll_wait(epollfd,events,events_num,tm.getRecentTimeout());
        tm.takeAllTimeout();

时间轮实现

另外一种常见的定时器设计使用的存放超时时间的容器叫做”时间轮”。微信的开源项目libco中使用的就是这种数据结构。

Hierarchy 时间轮的原理大致如下，下面是一个时分秒的Hierarchy时间轮，不同于Linux内核的实现，但原理类似。对于时分秒三级时间轮，每个时间轮都维护一个cursor，新建一个timer时，要挂在合适的格子，剩余轮数以及时间都要记录，到期判断超时并调整位置。原理图大致如下:

对于时间轮的实现，Timer依然是存放在链表上，但是借助了hash的思想，将相同间隔(或者相同周期的整数倍)的超时Timer放在同一个时间轮子上的槽(slot)上。时间轮上有一个指针，按照一个基准的频率(比如1ms,5ms,10ms等，libco中设置的是1ms)向前移动。这个基准的频率就是传递给epoll_wait()超时的值，也是定时器精度的基本单位。

class Timer
public:
    Timer(int rotations,int slot,std::function<void(void)> fun,void* args)
        : rotations_(rotations),slot_(slot),fun(fun) { }
    inline int getRotations() { return rotations_; }
    inline void decreaseRotations() { --rotations_; }
    inline void active() { fun(); }
    inline int getSlot() { return slot_; }
private:
    int rotations_;
    int slot_;
    std::function<void(void)> fun;
    void* args;

时间轮中的Timer类和最小堆中的实现，多了两个参数，rotations表示时间轮转多少圈后当前的Timer会触发，slot表示当前的Timer应该挂在对应的槽指向的链表上。成员函数比较简单，不多赘述。

class TimeWheel
  public:
    TimeWheel(int nslots)
        : nslosts_(nslots),curslot_(0),
            slots_(nslosts_,std::vector<Timer*>()),starttime_(getCurrentMillisecs()) { }
    ~TimeWheel();
    unsigned long long getCurrentMillisecs();
    Timer *addTimer(int timeout,std::function<void(void)> fun,void* args);
    void delTimer(Timer *timer);
    void tick();
    void takeAllTimeout();
private:
    int nslosts_;
    int curslot_;
    unsigned long long starttime_;
    std::vector<std::vector<Timer*>> slots_;
};

curslot_表示时间轮当前指向的那个slot。nslosts_表示这个时间轮拥有多少个slot，不要上图迷惑了，实际上slot会远远超过这8个。要想效率足够高，slot就要越大，当然占用的内存也会越大(现代计算机，这点内存可以忽略不计)，libco默认使用了 60 * 1000 个slot 。

addTimer()是添加一个Timer到TimeWheel上，需要根据传递的timeout参数，计算出该Timer所对应的slot (slot = (curslot_ + (timeout % nslosts_)) % nslosts_;) ,还有到这个Timer超时时间轮的指针需要转过的圈数(timeout / nslosts_)。

delTimer() 根据Timer*参数，删除时间轮上对应的Timer。

tick() 时间轮的指针走动一下。同时遍历当前slot上链表里的每一个Timer，如果Timer的圈数大于0，将Timer里的圈数-1，否则激活这个Timer。

takeAllTimeout() 是必要的，由于误差的存在，每次epoll_wait超时后(一个基准频率)，时间轮可能需要走动好几步。如果每次epoll_wait后直接tick()而不是takeAllTimeout()，会导致误差一直被累积，时间轮上剩余的定时器被滞后触发。

时间轮的实现我也放到了Github上: https:// gist.github.com/baixian gcpp/63278c0087201a655f940ab8de543abd 。使用方式和之前的最小堆实现的基本是一样的。

使用示例：

int dispatch()
    TimeWheel tw(60 * 1000);
    tw.addTimer(1000, []() { std::cout << "hello world" << std::endl; }, NULL);
    tw.addTimer(5000, []() { std::cout << "hello baixiancpp" << std::endl; }, NULL);
    for(;;)
        int ret = epoll_wait(epollfd,events,events_num,1); // 基准频率 1ms
        // tw.tick(); 不要这么做，会导致误差累积
        tw.takeAllTimeout();

最小堆和时间轮的时间复杂度，如下：

这里我只列出添加定时器和触发定时器的时间复杂度，因为这几种实现中，删除一个定时器可以优化到O(1)的时间复杂度————把其对应的回调函数置空。前面的例子我没有这么做，仅为展示，有兴趣的话可以自行修改。

乍看下来，时间轮的复杂度和链表是一样的。其实不然，时间轮上触发一个定时器，仅仅是理论上的O(n),只要slot的数量设置合理，时间复杂度会下降至接近O(1)。

可以根据实际需要，选择合适的定时器容器。

要不要用Timerfd？

开篇的表格里有提到，从Linux2.6.25开始，timerfd系列API，带来了一种全新的定时机制。把超时事件转换为了文件描述符，当超时发生后该文件描述符会变成可读。于是超时事件就变成了普通的IO事件。如果未对timerfd设置阻塞，对其read操作会一直阻塞到超时发生。此外timerfd的精度达到了纳秒级。不考虑跨平台等因素，这是一个非常不错的选择。

libevent2.1的源码里也支持timerfd了，在版本说明里也很明确了说明了使用多路复用的超时参数和使用timerfd之间的差异 ，它使用了两个词”efficient”和”precise”，分别表示这种实现之间的差异，我想着这还是非常有说服力的。

每个超时事件独享一个timerfd

如果对于每一个超时事件都用timerfd_create()创建一个对应的fd，放到epoll中统一管理。这样的做法是不合适的。每增加一个定时事件，都需要额外的3个系统调用:

此外，文件描述符还是稀缺的资源，每个进程能够使用的文件描述符是受系统限制的，如果定时器过多，会造成严重的浪费。

这种方式的定时器，比较容易实现，这里我就不再浪费篇幅了。

所有超时事件共享一个timerfd

libevent就是使用的这种方式。定时时间仍然使用最小堆来保存，每个event loop共享同一个timerfd。每次事件循环之前，取出最近的一个超时的时间，将这个timerfd设置为这个超时时间。

int epoll_dispatch( ...)
    if (epollop->timerfd >= 0)
        struct itimerspec is;
        is.it_value.tv_sec = tv->tv_sec;
        is.it_value.tv_nsec = tv->tv_usec * 1000;
        timerfd_settime(epollop->timerfd, 0, &is, NULL);
    res = epoll_wait(epollop->epfd, events, epollop->nevents, -1);
    for (i = 0; i < res; i++) 
        if (events[i].data.fd == epollop->timerfd)