本文共 5609 字，大约阅读时间需要 18 分钟。

定时器的实现原理

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

Linux内核定时器相关(Linux v4.9.7, x86体系架构)的一些相关代码：

内核启动注册时钟中断

// @file: arch/x86/kernel/time.c - Linux 4.9.7// 内核init阶段注册时钟中断处理函数static struct irqaction irq0  = {
       .handler = timer_interrupt,    .flags = IRQF_NOBALANCING | IRQF_IRQPOLL | IRQF_TIMER,    .name = "timer"};void __init setup_default_timer_irq(void){
       if (!nr_legacy_irqs())        return;    setup_irq(0, &irq0);}// Default timer interrupt handler for PIT/HPETstatic irqreturn_t timer_interrupt(int irq, void *dev_id){
       // 调用体系架构无关的时钟处理流程    global_clock_event->event_handler(global_clock_event);    return IRQ_HANDLED;}

内核时钟中断处理流程

// @file: kernel/time/timer.c - Linux 4.9.7/* * Called from the timer interrupt handler to charge one tick to the current * process.  user_tick is 1 if the tick is user time, 0 for system. */void update_process_times(int user_tick){
       struct task_struct *p = current;    /* Note: this timer irq context must be accounted for as well. */    account_process_tick(p, user_tick);    run_local_timers();    rcu_check_callbacks(user_tick);#ifdef CONFIG_IRQ_WORK    if (in_irq())        irq_work_tick();#endif    scheduler_tick();    run_posix_cpu_timers(p);}/* * Called by the local, per-CPU timer interrupt on SMP. */void run_local_timers(void){
       struct timer_base *base = this_cpu_ptr(&timer_bases[BASE_STD]);    hrtimer_run_queues();    /* Raise the softirq only if required. */    if (time_before(jiffies, base->clk)) {
           if (!IS_ENABLED(CONFIG_NO_HZ_COMMON) || !base->nohz_active)            return;        /* CPU is awake, so check the deferrable base. */        base++;        if (time_before(jiffies, base->clk))            return;    }    raise_softirq(TIMER_SOFTIRQ); // 标记一个软中断去处理所有到期的定时器}

内核定时器时间轮算法

单层时间轮算法的原理比较简单：用一个数组表示时间轮，每个时钟周期，时间轮 current 往后走一个格，并处理挂在这个格子的定时器链表，如果超时则进行超时动作处理，然后删除定时器，没有则剩余轮数减一。原理如图：

Linux 内核则采用的是 Hierarchy 时间轮算法，Hierarchy 时间轮将单一的 bucket 数组分成了几个不同的数组，每个数组表示不同的时间精度，Linux 内核中用 jiffies 记录时间，jiffies记录了系统启动以来经过了多少tick。下面是Linux 4.9的一些代码：

// @file: kernel/time/timer.c - Linux 4.9.7/* * The timer wheel has LVL_DEPTH array levels. Each level provides an array of * LVL_SIZE buckets. Each level is driven by its own clock and therefor each * level has a different granularity. *//* Size of each clock level */#define LVL_BITS    6#define LVL_SIZE    (1UL << LVL_BITS)/* Level depth */#if HZ > 100# define LVL_DEPTH  9# else# define LVL_DEPTH  8#endif#define WHEEL_SIZE  (LVL_SIZE * LVL_DEPTH)struct timer_base {
       spinlock_t      lock;    struct timer_list   *running_timer;    unsigned long       clk;    unsigned long       next_expiry;    unsigned int        cpu;    bool            migration_enabled;    bool            nohz_active;    bool            is_idle;    DECLARE_BITMAP(pending_map, WHEEL_SIZE);    struct hlist_head   vectors[WHEEL_SIZE];} ____cacheline_aligned;

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

定时器的使用方法

在Linux 用户空间程序开发中，常用的定期器可以分为两类：

1. 执行一次的单次定时器 single-short；
2. 循环执行的周期定时器 Repeating Timer；

其中，Repeating Timer 可以通过在Single-Shot Timer 终止之后，重新再注册到定时器系统里来实现。当一个进程需要使用大量定时器时，同样利用时间轮、最小堆或红黑树等结构来管理定时器。而时钟周期来源则需要借助系统调用，最终还是从时钟中断。Linux用户空间程序的定时器可用下面方法来实现：

• 通过alarm()或setitimer()系统调用，非阻塞异步，配合SIGALRM信号处理；
• 通过select()或nanosleep()系统调用，阻塞调用，往往需要新建一个线程；
• 通过timefd()调用，基于文件描述符，可以被用于 select/poll 的应用场景；
• 通过RTC机制, 利用系统硬件提供的Real Time Clock机制, 计时非常精确；

上面方法没提sleep()，因为Linux中并没有系统调用sleep()，sleep()是在库函数中实现，是通过调用alarm()来设定报警时间，调用sigsuspend()将进程挂起在信号SIGALARM上，而且sleep()也只能精确到秒级上，精度不行。当使用阻塞调用作为定时周期来源时，可以单独启一个线程用来管理所有定时器，当定时器超时的时候，向业务线程发送定时器消息即可。

一个基于时间轮的定时器简单实现

#include 
   
    #include 
    
     #include 
     
      #include 
      
       #define TIME_WHEEL_SIZE 8typedef void (*func)(int data);struct timer_node {    struct timer_node *next;    int rotation;    func proc;    int data;};struct timer_wheel {    struct timer_node *slot[TIME_WHEEL_SIZE];    int current;};struct timer_wheel timer = {
  {0}, 0};void tick(int signo){    // 使用二级指针删进行单链表的删除    struct timer_node **cur = &timer.slot[timer.current];    while (*cur) {        struct timer_node *curr = *cur;        if (curr->rotation > 0) {            curr->rotation--;            cur = &curr->next;        } else {            curr->proc(curr->data);  // bug-fix: 与下面一样交换位置            *cur = curr->next;            free(curr);        }    }    timer.current = (timer.current + 1) % TIME_WHEEL_SIZE;    alarm(1);}void add_timer(int len, func action){    int pos = (len + timer.current) % TIME_WHEEL_SIZE;    struct timer_node *node = malloc(sizeof(struct timer_node));    // 插入到对应格子的链表头部即可, O(1)复杂度    node->next = timer.slot[pos];    timer.slot[pos] = node;    node->rotation = len / TIME_WHEEL_SIZE;    node->data = 0;    node->proc = action;} // test case1: 1s循环定时器int g_sec = 0;void do_time1(int data){    printf("timer %s, %d\n", __FUNCTION__, g_sec++);    add_timer(1, do_time1);}// test case2: 2s单次定时器void do_time2(int data){    printf("timer %s\n", __FUNCTION__);}// test case3: 9s循环定时器void do_time9(int data){    printf("timer %s\n", __FUNCTION__);    add_timer(9, do_time9);}int main(){    signal(SIGALRM, tick);    alarm(1); // 1s的周期心跳    // test    add_timer(1, do_time1);    add_timer(2, do_time2);    add_timer(9, do_time9);    while(1) pause();    return 0;}
      
     
    
   

在实际项目中，一个常用的做法是新起一个线程，专门管理定时器，定时来源使用rtc、select等比较精确的来源，定时器超时后向主要的work线程发消息即可，或者使用timefd接口。

参考：