深入 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/documentation/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_;};

addTimer()参数和 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/baixiangcpp/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/baixiangcpp/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)            ;//    }}

这样的改进规避了前一种方式提到的造成文件描述符资源浪费的问题，仅仅需要 1 个额外的文件描述符。

额外的系统调用从额外的 3 个，降到了 1 个。而且还有改进的空间，只有当栈顶的 timeout 变化时，才调用 timerfd_settime()改变。

这种方式实现的定时器，精度提高了但是多了 1 个额外的系统调用。libevent 把选择权给了用户，用户可以根据实际情况在创建 event base 的时候是否配置 EVENT_BASE_FLAG_PRECISE_TIMER 宏而选择使用哪个定时器实现。

总结

std::priority_queue 是一个容器适配器，底层的容器默认使用的 std::vector(make_heap())。但是这不意味着往 std::priority_queue 插入一个元素的开销是 O(n)，C++标准对此实现有要求，可以放心大胆的去用。但是 std::priority_queue 没有提供高效删除元素的接口，我们可以通过将回调函数置空的方式，以 O(1)的时间复杂度实现删除。

以 C++实现的 muduo 网络库使用的是 std::set 集合存放 Timer：

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

实际上 std::set 实现应该是二叉搜索树，因此效率可能会比用 std::priority_queue 略差一点（《linux 多线程网络编程》 8.2 ）。

此外，libev 允许使用一个宏 EV_USE_4HEAP 指定以一个 4-heap 的数据结构保存定时器，据说效率更高，我也没有测试。

以上就是目前一些 c/c++语言实现的网络库里边定时器常用的设计手法。