简单描述时间轮_rocketmq 时间轮

大家好，又见面了，我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

时间轮

作用

也是用来作定时器触发任务，只是他更高效，时间复杂度为O(1)。

运行原理

为了方便理解我们参考钟表，它分为3个层次：时、分、秒，只有秒针在运动，走动一格时间为1秒，走一圈为1分钟，分针走一圈为1小时。
同样的，时间轮也分为多层，同样的只有第一层在运动，第一层走完，第二层走一格，第二层走完，第三次走一格，依次类推！！！
这样做就能用几个数组，代表一段较长的时间区间，左下图能计算0 – 135（ 5333 ），如果将层1数组长度设置为60，第二层设置为60，第三层设置12，第四层去掉，则就是右下图的时钟了，时间区间为 0 – 42200（6060*12）秒

时间与每层的索引关系

举个简单的4层时间轮例子（如左上图），假设最小计时单位为1（姑且理解为秒）
时间轮初始为0，那么给定任意时间time，求time落在每层时间轮的索引！
显然这是一个数学关系：
第一层：int first_index = time%5
第二层： int sencond_index = time/5 % 3
第三层： int third_index = time/5/3%3
第四层： int forth_index = time/5/3/3%3
如果不是很理解，可以举时钟的例子，假设当前时间为 3820 秒，求时分秒指针在哪个位置？ 显然：
秒：3820%60 = 40
分：3820/60%60 = 3；
时：3820/60/60%12 = 1；

如何把事件与时间轮结合？

从数据结构设计

时间轮是由多个定长数组组成的，我们只需要把事件接在数组中就可以了，由于同一时刻会有多个事件，考虑先添加的事件先执行，使用链表来把事件连接起来，因此时间轮是一个 定长的包含链表的数组

事件添加过程

只要知道了时间，就能得到给定的时间所在的层数与索引，就能找到该位置对应的事件链表，然后把事件添加进去

执行过程

为了方便说明，依然以下图为例，以time作为从0开始时间，单位为秒，每过一秒 time+1。
事件执行永远只在1层，每过1秒数组索引右移一位，同时执行该数组里面的事件，执行完后删除事件，完整遍历后，需要把2层的事件更新到1层来执行，2层的事件都执行完后，需要把3层的事件更新到1、2，依此类推，直到最后一层。事件更新后又回到起始位置进行新一轮的事件执行过程。
显然，任务执行的过程很容易理解，用time对5求余就能知道事件执行到了那一刻，那么何时分发呢？注意看下图标红部分，因为举的例子层数和每层的数组长度都很少，先不考虑第四次，很容易知道当time等于 5、10、15、20、25、30、35、40、45时需要先分发事件在执行，不难看出：
time% 5 == 0 时需要找把2层的事件更新到1层 （5）
time%15 == 0 时需要找把3层的事件更新到2、1层 （53）
time%45 == 0 时需要找把4层的事件更新到3、2、1层 （53*3）
…
**注意：**时间轮会有自己的最大计时区间，区间范围取决于时间轮层数及每层数组的大小，下图只有135秒的计时范围

实现过程

以我自己的demo为例。
数据结构：
1、定义任务节点，组成任务链，节点应该包含需要执行的任务和任务执行的时间（以时间轮为起始点的时间）
2、定长链表数组，组成多层轮子，链表的节点为1定义的节点
3、定义时间变量，记录时间轮从起始时刻到当前的时间
成员函数：
1、增加节点函数
2、任务执行函数
3、数据更新函数（主要是为了从下层时间轮拿任务节点）
源代码有注释
头文件

//任务节点
class Node{ 
   
public:
    qint64 expire = 0;
    Node *next = nullptr;
};
//任务节点链表
class NodeList{ 
   
public:
    Node head;
    Node *tail = nullptr;
};
class TimeWheel
{ 
   
public:
    TimeWheel();

    const static qint64 FIRST_MAST = 8;
    const static qint64 EACH_MAST  = 6;
    const static qint64 WHEEL_LEVEL = 5;
    const static qint64 v = 1;
    const static qint64 WHEEL_EACH_NUM = v<<EACH_MAST;
    const static qint64 WHEEL_FIRST_NUM = (v<<FIRST_MAST);

    //定长链表数组
    NodeList first_wheel[WHEEL_FIRST_NUM];
    NodeList other_level_wheel[WHEEL_LEVEL-1][WHEEL_EACH_NUM];

    void addNodeInList(NodeList&,Node*node);
    Node* clearNodeList(NodeList&);
    void dispatchNode(NodeList&);
    qint64 curtime;//单位为时间轮的单位刻度

    void checkWheelUpdate();//数据更新函数
    void addNode(Node *node);//添加节点函数
    void addNode(int expire);
    void execScale();
    void execTask();//任务执行函数


};

#endif // TIMEWHEEL_H

源文件

#include "timewheel.h"
#include <QDebug>
TimeWheel::TimeWheel()
{ 
   
    curtime = 0;

    qDebug()<<"WHEEL_FIRST_NUM:"<<WHEEL_FIRST_NUM;
    qDebug()<<"WHEEL_EACH_NUM:"<<WHEEL_EACH_NUM;
}

void TimeWheel::addNodeInList(NodeList&list,Node*node)
{ 
   

    if(list.head.next == nullptr){ 
   
        list.head.next = node;
        list.tail = node;
    }else{ 
   

        list.tail->next = node;
        list.tail = node;
    }

}

void TimeWheel::addNode(int expire)
{ 
   
    Node *node = new Node;
    node->expire = curtime + expire;
    node->next = nullptr;
    addNode(node);

}

/* 注意节点的expire时间用来确认事件应该放在数组的哪个位置 delay 用来判断在哪层时间轮 */
void TimeWheel::addNode(Node *node)
{ 
   

    int t = node->expire;
    int delay = node->expire - curtime;
    int index = 0;
    if(delay < WHEEL_FIRST_NUM){ 
   

        index = t % WHEEL_FIRST_NUM;
        addNodeInList(first_wheel[index],node);

    }else if(delay < WHEEL_EACH_NUM * WHEEL_FIRST_NUM){ 
   

        index = t/ WHEEL_FIRST_NUM%WHEEL_EACH_NUM;
        addNodeInList(other_level_wheel[0][index],node);

    }else if(delay < WHEEL_EACH_NUM*WHEEL_EACH_NUM * WHEEL_FIRST_NUM){ 
   

        index = t/ (WHEEL_EACH_NUM * WHEEL_FIRST_NUM)%WHEEL_EACH_NUM;
        addNodeInList(other_level_wheel[1][index],node);

    }else if(delay < WHEEL_EACH_NUM * WHEEL_FIRST_NUM){ 
   

        index = t/ (WHEEL_EACH_NUM *WHEEL_EACH_NUM * WHEEL_FIRST_NUM)%WHEEL_EACH_NUM;
        addNodeInList(other_level_wheel[2][index],node);

    }else if(delay < WHEEL_EACH_NUM * WHEEL_FIRST_NUM){ 
   

        index = t/ (WHEEL_EACH_NUM *WHEEL_EACH_NUM *WHEEL_EACH_NUM * WHEEL_FIRST_NUM)%WHEEL_EACH_NUM;
        addNodeInList(other_level_wheel[3][index],node);

    }




}

void TimeWheel::dispatchNode(NodeList& list)
{ 
   


    Node *curnode = list.head.next;

    //将节点分发时，需要清楚当前链表上的节点
    list.head.next = nullptr;
    list.tail = &list.head;

    while (curnode != nullptr) { 
   

        Node *tmpnode = curnode;
        curnode = curnode->next;

        //因为添加节点时要保证节点为1个，而不是一个链表
        //固将链表的连接点打断
        tmpnode->next = nullptr;
        addNode(tmpnode);
    }
}

Node* TimeWheel::clearNodeList(NodeList&list)
{ 
   

    Node*node = list.head.next;

    list.head.next = nullptr;
    list.tail = &list.head;
    return node;
}

void TimeWheel::execTask(){ 
   

    int index = curtime % WHEEL_FIRST_NUM;

    Node *curnode = clearNodeList(first_wheel[index]);

    while (curnode != nullptr) { 
   

        //执行定时节点任务
        qDebug()<<"expire time:"<<curnode->expire;
        Node *nextnode = curnode->next;
        delete curnode;
        curnode =  nextnode;

    }

}

void TimeWheel::execScale()
{ 
   

    execTask();//执行任务
    checkWheelUpdate();//检测更新
    curtime++;//时间刻度更新
}


void TimeWheel::checkWheelUpdate()
{ 
   
    qint64 dispatch_radio = WHEEL_FIRST_NUM;
    qint64 dispatch_mask = curtime % dispatch_radio;

    qint64 tmpa = WHEEL_FIRST_NUM;

    int level = 0;
    while(curtime != 0 && dispatch_mask == 0){ 
   

        dispatch_radio *= WHEEL_EACH_NUM;
        dispatch_mask = curtime % dispatch_radio ;
        if(dispatch_mask == 0 ){ 
   
            //如果整除，则往下一层找
            level++;
            tmpa *= WHEEL_EACH_NUM;
            continue;
        }

        int index = curtime/tmpa%WHEEL_EACH_NUM;
        dispatchNode(other_level_wheel[level][index]);
    }
}

由于时间轮每层的数组的长度可以定义为任何值，并且时间轮涉及到了很多乘法和除法和取余，所以可以考虑使用位运算来替代运行。

完整Demo源码

qt c++代码