定時(shí)器作用及實(shí)現(xiàn)定時(shí)器數(shù)據(jù)結(jié)構(gòu)選取介紹1

本文主要介紹定時(shí)器作用，實(shí)現(xiàn)定時(shí)器數(shù)據(jù)結(jié)構(gòu)選取，并詳細(xì)介紹了跳表，紅黑樹(shù)，時(shí)間輪實(shí)現(xiàn)定時(shí)器的思路和方法。

”

定時(shí)器作用

定時(shí)器在各種場(chǎng)景都需要用到，比如游戲的Buff實(shí)現(xiàn)，Redis中的過(guò)期任務(wù)，Linux中的定時(shí)任務(wù)等等。顧名思義，定時(shí)器的主要用途是執(zhí)行定時(shí)任務(wù)。

定時(shí)器數(shù)據(jù)結(jié)構(gòu)選取

定時(shí)器數(shù)據(jù)結(jié)構(gòu)要求：

• 需要快速找到到期任務(wù)，因此，應(yīng)該具有有序性；
• 其過(guò)期執(zhí)行、插入(添加定時(shí)任務(wù))和刪除(取消定時(shí)任務(wù))的頻率比較高，三種操作效率必須保證

以下為各數(shù)據(jù)結(jié)構(gòu)時(shí)間復(fù)雜度表現(xiàn)

有序鏈表：插入O(n)，刪除O(1)，過(guò)期expire執(zhí)行O(1)

最小堆：插入O(logn)，刪除O(logn)，過(guò)期expire執(zhí)行O(1)

紅黑樹(shù)：插入O(logn)，刪除O(logn)，過(guò)期expire執(zhí)行O(logn)

哈希表+鏈表（時(shí)間輪）：插入O(1)，刪除O(1)，過(guò)期expire平均執(zhí)行O(1)（最壞為O(n)）

不同開(kāi)源框架定時(shí)器實(shí)現(xiàn)方式不一，如，libuv采用最小堆來(lái)實(shí)現(xiàn)，nginx采用紅黑樹(shù)實(shí)現(xiàn)，linux內(nèi)核和skynet采用時(shí)間輪算法實(shí)現(xiàn)等等。

定時(shí)器接口封裝

作為定時(shí)器，需要封裝以下4類(lèi)接口給用戶使用：

• 創(chuàng)建定時(shí)器：init_timer
• 添加定時(shí)任務(wù)：add_timer
• 取消定時(shí)任務(wù)：cancel_timer
• 執(zhí)行到期任務(wù)：expire_timer

其中執(zhí)行到期任務(wù)有兩種工作方式：

1. 輪詢(xún): 每隔一個(gè)時(shí)間片去查找哪些任務(wù)到期
2. 睡眠/喚醒：不停查找deadline最近任務(wù)，到期執(zhí)行，否則sleep；sleep期間，任務(wù)有改變，線程會(huì)被喚醒

接下來(lái)將介紹分別用跳表、紅黑樹(shù)、時(shí)間輪來(lái)實(shí)現(xiàn)定時(shí)器。

跳表實(shí)現(xiàn)定時(shí)器

跳表簡(jiǎn)介

跳表是一種動(dòng)態(tài)的數(shù)據(jù)結(jié)構(gòu)，采用空間換時(shí)間的思想，在有序鏈表基礎(chǔ)上加入多級(jí)索引，通過(guò)索引進(jìn)行二分快速查找，支持快速刪除、插入和查找操作（平均時(shí)間復(fù)雜度為O(logN)，最壞為O(N)），效率可與平衡樹(shù)媲美，實(shí)現(xiàn)比其簡(jiǎn)單。

下面通過(guò)一張圖來(lái)簡(jiǎn)單說(shuō)明跳表操作。跳表的最底層即為基本的有序鏈表，存儲(chǔ)所有的數(shù)據(jù)，可理解為數(shù)據(jù)層；往上則為索引層，理想狀態(tài)下，上一層為下一層節(jié)點(diǎn)數(shù)的一半。比如，要查找下圖的數(shù)據(jù)為11的節(jié)點(diǎn)，從begin''出發(fā)，向右走，如果下一個(gè)節(jié)點(diǎn)大于11則往下走，直到找到目標(biāo)節(jié)點(diǎn)?？梢?jiàn)，跳表要比原始鏈表少比較一些節(jié)點(diǎn)，但前提是需要花更多空間存儲(chǔ)索引節(jié)點(diǎn)。

image-20210323182236910

跳表實(shí)現(xiàn)定時(shí)器

• 跳表查找，插入，刪除(任意節(jié)點(diǎn)、頭節(jié)點(diǎn))的時(shí)間復(fù)雜度大概率趨向于O(logn)
• 過(guò)期任務(wù)查找，只需要跟第一個(gè)節(jié)點(diǎn)比較，因其第一個(gè)節(jié)點(diǎn)即為最小節(jié)點(diǎn)

學(xué)會(huì)吸取開(kāi)源框架中優(yōu)秀數(shù)據(jù)結(jié)構(gòu)和代碼思想，直接采用redis中跳表結(jié)構(gòu)的實(shí)現(xiàn)，取出所需部分，用于實(shí)現(xiàn)定時(shí)器。如下：

跳表數(shù)據(jù)結(jié)構(gòu)

跳表節(jié)點(diǎn)與跳表結(jié)構(gòu)

/*skiplist.h*/
#define ZSKIPLIST_MAXLEVEL 32
#define ZSKIPPLIST 0.25

typedef struct zskiplistNode zskiplistNode;
typedef void (*handler_pt) (zskiplistNode * node);
// 跳表節(jié)點(diǎn)
struct zskiplistNode {
  unsigned long score;  /*用于排序的值*/
  handler_pt handler;  /*處理函數(shù)*/
  struct zskiplistLevel {
    struct zskiplistNode **forward;
  }level[];
};
// 跳表結(jié)構(gòu)
typedef struct zskiplist {
  struct zskiplistNode * header;
  int length;
  int level;  /*跳表層數(shù)*/
}zskiplist;

跳表接口申明

具體接口實(shí)現(xiàn)細(xì)節(jié)請(qǐng)移步redis源碼。

/*skiplist.h*/
/*創(chuàng)建跳表，初始化*/
zskiplist *zslCreate(void);
/*刪除跳，表釋放資源*/
void zslFree(zskiplist *zsl);
/*插入節(jié)點(diǎn)*/
zskiplistNode *zslInsert(zskiplist *zsl, unsigned long score, handler_pt func);
/*刪除頭節(jié)點(diǎn)*/
void zsklDeleteHead(zskiplist *zsl);
/*刪除任意節(jié)點(diǎn)*/
void zslDelete(zskiplist *zsl, zskplistNode *zn);
/*打印，調(diào)試*/
void zslPrint(zskiplist *zsl);

定時(shí)器接口實(shí)現(xiàn)

主要介紹四個(gè)接口實(shí)現(xiàn)：初始化定時(shí)器，添加定時(shí)任務(wù)，刪除/取消定時(shí)任務(wù)，處理定時(shí)任務(wù)

// test_user.c  封裝給用戶使用的接口
static uint32_t
current_time() {
 uint32_t t;
    struct timespec ti;
    clock_getttime(CLOCK_MONOTONIC, &ti);
    t = (uint32_t)ti.tv_sec * 1000;
    t += ti.tv_sec / 1000000;
}
zskiplist *init_timer() {
    // 初始化定時(shí)器
    return zslCreate();
}
zskiplistNode *add_timer(zskiplist *zsl, uint32_t msec, handler_pt func) {
    // 添加定時(shí)任務(wù)
    msec += current_time();
    return zslInsert(zsl, msec, func);
}
void cancel_timer(zskiplist *zsl, zskiplistNode *zn) {
    // 刪除/取消定時(shí)任務(wù)
    zslDelete(zsl, zn);
}
void expire_timer(zskiplist *zsl){
    // 處理定時(shí)任務(wù)
    zskiplistNode *x;
    uint32_t now = current_time();
    for (;;) {
        x = zslMin(zsl);  // 最近節(jié)點(diǎn)
        if (!x) break;
        if (x->score > now)  break;  // 時(shí)間未到
        x->handler(x);  // 執(zhí)行相關(guān)定時(shí)任務(wù)
        zslDeleteHead(zsl);  // 執(zhí)行完刪除
    }
}

紅黑樹(shù)實(shí)現(xiàn)定時(shí)器

紅黑樹(shù)

紅黑樹(shù)是一種自平衡的二叉查找樹(shù)，即，插入和刪除操作如果破壞樹(shù)的平衡時(shí)，需要重新調(diào)整達(dá)到平衡狀態(tài)。因此，是一種比較難的數(shù)據(jù)結(jié)構(gòu)。

紅黑樹(shù)五條性質(zhì)

• 每個(gè)節(jié)點(diǎn)要么是紅色，要么是黑色
• 根節(jié)點(diǎn)是黑色
• 每個(gè)葉子結(jié)點(diǎn)是黑色
• 每個(gè)紅節(jié)點(diǎn)的兩個(gè)子節(jié)點(diǎn)一定是黑色
• 任意一節(jié)點(diǎn)到每個(gè)葉子節(jié)點(diǎn)的路徑都含相同數(shù)目的黑結(jié)點(diǎn)

弄懂紅黑樹(shù)如何調(diào)整樹(shù)的平衡，保證滿足這5條性質(zhì)，是比較麻煩，需要耐心的去推導(dǎo)一遍，此處不展開(kāi)。

紅黑樹(shù)實(shí)現(xiàn)定時(shí)器

AVL 樹(shù)平衡要求太高，維護(hù)平衡操作過(guò)多，較復(fù)雜；紅黑樹(shù)只需維護(hù)一個(gè)黑高度，效率較高

紅黑樹(shù)查找，刪除，添加時(shí)間復(fù)雜度為：O(log(n))

吸取開(kāi)源框架中優(yōu)秀數(shù)據(jù)結(jié)構(gòu)和代碼思想，選用nginx中的紅黑樹(shù)結(jié)構(gòu)

紅黑樹(shù)數(shù)據(jù)結(jié)構(gòu)

紅黑樹(shù)節(jié)點(diǎn)與紅黑樹(shù)

// rbtree.h  紅黑樹(shù)數(shù)據(jù)結(jié)構(gòu)以及相關(guān)接口，具體接口實(shí)現(xiàn)同上
#ifndef _NGX_RBTREE_H_INCLUDE_
#define _NGX_RBTREE_H_INCLUDE_

typedef unsigned int ngx_rbtree_key_t;
typedef unsigned int ngx_uint_t;
typedef int ngx_rbtree_key_int_t;

// 紅黑樹(shù)節(jié)點(diǎn)
typedef struct ngx_rbtree_node_s  ngx_rbtree_node_t;
struct ngx_rbtree_node_s {
    ngx_rbtree_key_t key;
    ngx_rbtree_node_t *left;
    ngx_rbtree_node_t *right;
    ngx_rbtree_node_t *parent;
    u_char    color;  // 節(jié)點(diǎn)顏色
    u_char    data;  // 節(jié)點(diǎn)數(shù)據(jù)
};
// 插入函數(shù)指針
typedef void (*ngx_rbtree_insert_pt) (ngx_rbtree_node_t *root,
     ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel);
// 紅黑樹(shù)
typedef struct ngx_rbtree_s ngx_rbtree_t;
struct ngx_rbtree_s {
    ngx_rbtree_node_t  *root;
    ngx_rbtree_node_t  *sentinel;
    ngx_rbtree_insert_pt insert;
};

紅黑樹(shù)接口聲明

// 紅黑樹(shù)初始化
#define ngx_rbtree_init(tree, s, i)       \\
 ngx_rbtree_sentinel_init(s);      \\
 (tree)->root = s;        \\
 (tree)->sentinel = s;       \\
 (tree)->insert = i;        
// 插入操作
void ngx_rbtree_insert(ngx_rbtree_t *tree, ngx_rbtree_node_t *node);
// 刪除操作
void ngx_rbtree_delete(ngx_rbtree_t *tree, ngx_rbtree_node_t *node);
// 插入value
void ngx_rbtree_insert_value(ngx_rbtree_node_t *root, ngx_rbtree_node_t *node,
                            ngx_rbtree_node_t *sentinel);
// 插入timer
void ngx_rbtree_insert_timer_value(ngx_rbtree_node_t *root,
                                  ngx_rbtree_node_t *node,
                                  ngx_rbtree_node_t *sentinel);
// 獲取下一個(gè)節(jié)點(diǎn)
ngx_rbtree_node_t *ngx_rbtree_next(ngx_rbtree_t *tree, ngx_rbtree_node_t *node);
#define ngx_rbt_red(node)    ((node)->color = 1)
#define ngx_rbt_black(node)    ((node)->color = 0)
#define ngx_rbt_is_red(node)   ((node)->color)
#define ngx_rbt_is_black(node)   (!ngx_rbt_is_red(node))
#define ngx_rbt_copy_color(n1, n2)  (n1->color = n2->color)
#define ngx_rbtree_sentinel_init(node)  ngx_rbt_black(node)
// 找到最小值，一直往左走即可
static inline ngx_rbtree_node_t *
ngx_rbtree_min(ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel)
{
    while (node->left != sentinel){
        node = node->left;
    }
    return node;
}

定時(shí)器接口實(shí)現(xiàn)

// test_user.c  封裝給用戶使用的接口
ngx_rbtree_t     timer;
static ngx_rbtree_node_t   sentinel;
typedef struct timer_entry_s timer_entry_t;
typedef void (*timer_handler_pt)(timer_entry_t *ev);

struct timer_entry_s {
   ngx_rbtree_node_t timer;
    timer_handler_pt  handler;
};
// 初始化
int init_timer() {
    ngx_rbtree_init(&timer, &sentinel, ngx_rbtree_insert_timer_value);
    return 0;
}
// 添加定時(shí)任務(wù)
void add_timer(timer_entry_t *te, uint32_t msec) {
    msec += current_time();
    te->timer.key = msec;
    ngx_rbtree_insert(&timer, &te->timer);
}
// 取消定時(shí)
void cancel_timer(timer_entry_t *te) {
    ngx_rbtree_delete(&timer, &te->timer);
}
// 執(zhí)行到期任務(wù)
void expire_timer() {
    timer_entry_t *te;
    ngx_rbtree_node_t *sentinel, *root, *node;
    sentinel = timer.sentinel;
    uint32_t now = current_time();
    for(;;){
        root = timer.root;
        if (root == sentinel) break;
        if (node->key > now) break;
        te = (timer_entry_t *) ((char *) node - offsetof(timer_entry_t, timer));
        te->handler(te);
        ngx_rbtree_delete(&timer, &te->timer);
        free(te);
    }
}

以上，為紅黑樹(shù)和跳表實(shí)現(xiàn)的定時(shí)器，多線程環(huán)境下加鎖粒度比較大，高并發(fā)場(chǎng)景下效率不高，而時(shí)間輪適合高并發(fā)場(chǎng)景，如下。

時(shí)間輪實(shí)現(xiàn)定時(shí)器

時(shí)間輪

可以用于高效的執(zhí)行大量定時(shí)任務(wù)，如下為分層時(shí)間輪示意圖：

timewheel

時(shí)間輪可參考時(shí)鐘進(jìn)行理解，秒針(Seconds wheel)轉(zhuǎn)一圈，則分針(Minutes wheel)走一格，分針(Minutes wheel)轉(zhuǎn)一圈，則時(shí)針(Hours wheel)走一格。隨著，時(shí)間的流逝，任務(wù)不斷從上層流下下一層，最終到達(dá)秒針輪上，當(dāng)秒針走到時(shí)執(zhí)行。

如上所示，時(shí)間輪大小為8格，秒針1s轉(zhuǎn)動(dòng)一格，其每一格所指向的鏈表保存著待執(zhí)行任務(wù)。比如，如果當(dāng)前指針指向1，要添加一個(gè)3s后執(zhí)行的任務(wù)，由于1+3=4，即在第4格的鏈表中添加一個(gè)任務(wù)節(jié)點(diǎn)即可。如果要添加一個(gè)10s后執(zhí)行的任務(wù)，10+1=11，超過(guò)了秒針輪范圍，因此需要對(duì)8取模11 % 8 = 3，即，會(huì)把這個(gè)任務(wù)放到分針輪上3對(duì)應(yīng)的鏈表上，之后再?gòu)姆轴樰啺讶蝿?wù)丟到秒針輪上進(jìn)行處理。也即，**秒針輪(Seconds wheel)**即保存著最近將要執(zhí)行的任務(wù)，隨著時(shí)間的流逝，任務(wù)會(huì)慢慢的從上層流到秒針輪中進(jìn)行執(zhí)行。

優(yōu)點(diǎn)：加鎖粒度較小，只需要加一個(gè)格子即可，一個(gè)格子對(duì)應(yīng)一串鏈表；適合高并發(fā)場(chǎng)景

缺點(diǎn)：不好刪除

如何解決時(shí)間輪定時(shí)任務(wù)刪除？

1. 通過(guò)引用計(jì)數(shù)來(lái)解決
2. 交由業(yè)務(wù)層處理，將刪除標(biāo)記設(shè)為true , 在函數(shù)回調(diào)中根據(jù)這個(gè)標(biāo)記判斷是否需要處理

這里介紹兩種定時(shí)器實(shí)現(xiàn)方案，一種是簡(jiǎn)單實(shí)現(xiàn)方案，另一種是skynet較為復(fù)雜的實(shí)現(xiàn)。

時(shí)間輪實(shí)現(xiàn)定時(shí)器

簡(jiǎn)單時(shí)間輪實(shí)現(xiàn)方案

功能場(chǎng)景：由心跳包進(jìn)行超時(shí)連接檢測(cè)，10s未收到則斷開(kāi)連接

一般做法：map每秒輪詢(xún)這個(gè)結(jié)構(gòu)，檢測(cè)所有連接是否超時(shí)，收到心跳包，記錄時(shí)間戳

缺點(diǎn)：效率很差，每次需要檢測(cè)所有連接，時(shí)間復(fù)雜度為O(n)

優(yōu)化：分治大法，只需檢測(cè)快過(guò)期的連接， 采用hash數(shù)組+鏈表形式，數(shù)組大小設(shè)置成16 ：[0] + [1] + [2] + ... + [15] ，相同過(guò)期時(shí)間的放入一個(gè)數(shù)組，因此，每次只需檢測(cè)最近過(guò)期的數(shù)組即可，不需要遍歷所有。

數(shù)據(jù)結(jié)構(gòu)定義

以下為定時(shí)器節(jié)點(diǎn)，增加引用計(jì)數(shù)ref， 只有當(dāng)ref為0時(shí)刪除連接。

class CTimerNode {
public:
    CTimerNode(int fd) : id(fd), ref(0) {}
    void Offline() {this->ref = 0};
    bool tryKill() {
        if (this->ref == 0) return true;
        DecRef();
        if (this->ref == 0){
            return true;
        }
        return false;
    }
    void IncRef() {this->ref++;}
protected:
    void DecRef() {this->ref--;}
private:
    int ref;
    int id;
}
// 時(shí)間輪數(shù)組大小16, (x對(duì)16取余)==(x&1111) 落到0-15之間,即落到對(duì)應(yīng)的數(shù)組
const int TW_SIZE = 16;
const in EXPIRE = 10; // 過(guò)期間隔
const int TW_MASK = TW_SIZE - 1;  // 掩碼, 用于對(duì)16取余
static size_t iReadTick = 0;  // 滴答時(shí)鐘
typedef list

定時(shí)器接口

// 添加定時(shí)
void AddTimeOut(TimerWheel &tw, CTimerNode *p) {
    if (p) {
        p->IncRef();
        // 找到iRealTick對(duì)應(yīng)數(shù)組的idx(槽位)
        TimeList &le = tw[(iRealTick+EXPIRE) & TW_MASK];
        le.push_back(p);  // 把時(shí)間節(jié)點(diǎn)加入list中
    }
}
// 延時(shí)調(diào)用
void AddTimeOutDelay(TimeWheel &tw, CTimerNode *p, size_t delay) {
    if (p) {
        p->IncRef();
        TimeList &le = tw[(iRealTick + EXPIRE + delay) & TW_MASK];
        le.push_back(p);
    }
}
// 時(shí)間輪移動(dòng)
void TimerShift(TimeWheel &tw) {
    size_t tick = iRealTick;
    iRealTick++;
    TimeList &le = tw[tick & TW_MASK];
    TimeListIter iter = le.begin();
    for (; iter != le.end(); iter++) {
        CTimerNode *p = *iter;
        if (p && p->trySkill()){
            delete p;
        }
    }
    le.clear();
}