当前位置：首页 > news >正文

C++ 定时器

news 2025/6/27 6:09:20

这是第一次独立设计一个模块，从接口定义，模块组合到多线程并发可能遇到的各种问题，虽然定时挺简单的，但是想设计精度高，并且能应对高并发似乎也不是很容易，当然，最后没有测试定时器的代码，不知道能到什么水平，只是想记录整个过程中遇到的问题，和一些思考。

一、模块构成

其是一开始就想把整个定时器分成三个模块：

ticker：负责每一次滴答的，就是秒针。
taskstruct：用来管理任务，能够插入、删除任务。
wokerpool：同来执行任务的线程(俗称牛马)。
timer：这个定时器模块。

这就是像的最初结构。

1.1 TaskStruct

taskstruct用来管理各个任务。我第一个想到的就是时间轮算法，timewheel。

大概就是这样子(画得怪丑的.....)，插入任务，需要先找到是哪个盘子，然后是哪个slot，插入任务队列。这样面临两个问题

如何快速的定位到wheel和slot？
用什么数据管理每个slot的task？

第一个问题：关系到插入的效率。第二个问题：因为当第二个wheel指针移动到下一个slot时，需要将指针指向的slot队列移动到前一个wheel，这个过程如何高效；并且新来的如何如何快速插入到slot中。

对于第一个问题

采用bitmap的方式快速定位，怎么理解呢？首先linux中时间的精度一般是微妙(当然可以达到纳秒，但是纳秒似乎精度太高了，程序的运行会导致相对误差太大，暂且先用微妙)。通过gettimeofday函数正好获得的也是微妙级。换算过来就是uint64位的数字，表示1900年哒哒哒的东西(具体查资料吧，之前也写过linux的时间库)，然后定时时间加入是100微妙，用uint64表示就是

0110 0100

如图中所示，加入每个step是1微妙，每个wheel有8个slot，从0到7；那么第1个wheel可以表示的时间就是

0～7
000 000 000 ～ 000 000 111

这意味着可以用bitmap区索引wheel，和slot，最高位在最后三个bit中，就是在第一个；最高位在中间三个bit中，就是第二个wheel；最高位在前三个bit中，就是在第三个wheel。这个这三个bit的序号就是slot的索引。这样划分bit就能快速定位wheel和slot。

对于第二个问题

用什么样的数据结构slot的数据，最开始我有四个备选项：链表、有序数组、红黑树、跳表。

链表：可以在O(1)时间插入删除，但是查找时间需要O(n)，但是从后一个wheel移动到前一个wheel只要O(1)的时间，只要接在最后就行了。
有序数组：查找时间是O(log n)，但是删除和插入的时间都是O(n)，因为要移动数据。从后一个wheel移动到前一个wheel相当于一次查找和插入，因为是有序的，应该是O(n^2)(不知道对不对....)反正总之，有序数组效率很低。
红黑树：这个数据结构其是不错，但是红黑树的插入删除涉及到树的旋转(虽然不会超过三次，即记得好像是，但是需要从低到高递归变色，插入大概有三种情况，删除分为5种吧，记不清了)关于树的旋转也觉得不是很快。
跳表：我觉得这个是相对比较均衡的数据结构，有序的链表，插入删除都是logn，合并也是O(n)。并且，想leveldb底层数据不也是跳表吗。

但是，我总觉得，定时器需要考虑大量的删除操作吗？定时任务到期完全可以通过相关状态判断任务是否需要执行，就拿raft算法来说，leader每次都会发送心跳包，而follower收到心跳数据，就会重置计时器，重置计时器要么变为查找，删除，查找，插入操作。其实完全没必要搞这么麻烦，每个follower在变成follower时，记录一个原子变量uint64 i = 0；然后每次添加定时器的时候把i值给定时器，相当于给个票，每次收到心跳，i原子自增，定时器到期后对比拿到的号和i一样吗，一样，表示这段时间没有收到心跳数据，不一样表示收到了。所以，我觉得完全没必要删除定时器，可以根据一个状态位判断任务是否执行。顺便计算一下，一个64位i最大表示是10^19，按心跳包间隔1ns，还需要200年才到期，而这是整个系统完全没有leader和follower变动一直两百年。所以完全不需要担心回到起点的问题。

其次，实际上链表数据结构在slot上是无序的，但是，当一个slot被放到前一个wheel上是，就是一种排序(这大概就是归并排序的思想)随着不断往前，原本的一个slot数据会不断排序，wheel越靠前，slot之间的间隔越小，在第一个wheel上的slot是同一时间的，所以，删除操作还有另一个巧妙的方法，就是在做一定时器，这个定时器只比需要删除的早一点，当他到期时，就执行删除操作，通过时间定位到在第一个wheel的哪个slot，直接删除会快很多，但是这个带来的问题就是第一个wheel上的并发竞争很大，因为第一个wheel是秒针，秒针会很快的转动，要执行任务，还要执行删除，感觉还是不是很合适。

经过上面的分析，决定使用链表，认为不需要删除任务，将任务到期的执行逻辑交给上层应用保证。最终TimerStruct接口如下：

typedef std::function<void()> TimeDelayTask;struct TimerEntry{Timestamp m_oTimeStamp;TimeDelayTask m_fTask;explicit TimerEntry(Timestamp& timestamp, TimeDelayTask task): m_oTimeStamp(timestamp),m_fTask(task){}virtual ~TimerEntry() {}void SetTimestamp(Timestamp timestamp) { m_oTimeStamp = timestamp; }void SetTask(TimeDelayTask task) { m_fTask = std::move(task); }void SetTask(TimeDelayTask&& task) { m_fTask = std::move(task); }Timestamp GetTimestamp() { return m_oTimeStamp; }const TimeDelayTask GetTask() { return m_fTask; }
};class TimerStruct{
public:virtual bool DeleteEntry(Timestamp& timestamp) = 0; virtual bool InsertEntry(Timestamp& timestamp, TimeDelayTask& task) = 0;virtual bool Start() = 0;virtual bool Stop() = 0;virtual void Tick() = 0;virtual bool Clear() = 0;virtual bool Reset(Timestamp& tmstamp) = 0;virtual bool IsRunning() = 0;void RegistToTimer(Timer* tm);void UnregistToTimer();virtual uint64_t GetTaskNum() = 0;virtual ~TimerStruct() {}
protected:Timestamp m_uiStartTime;Timer* m_opTimer;uint64_t m_uiInterval;};typedef std::list<TimerEntry*> TimeDelayTaskQueue;

TimerEntry是一个任务，包括一个时间戳和延迟执行的任务。TimerStruct是管理任务的接口类。提供插入、删除、开始、停止、清空、重置、以及挂载到timer上的接口(虽然我认为删除操作是没有必要的，但是，接口还是留着吧，还有这个start和stop和isrunning，实现的时候，我发现好像也不是很需要，计时器的开始与停止完全可以有timer控制，但是，我总觉得，这个东西还是有必要留一下.....)。

1.1.1 timewheel

wheel两两之间通过双向链表连接，没有写一个control去控制所有的wheel，addtask时，先判断是否在第一个wheel上，不在就进入下一个，判断是否早下一个，递归进入。这里其实在实现过程中，我发现这种方式不是很好，应该一个controler用来管理每一个wheel，这样效率会更高点，但是代码没有做(有点懒，后面看机会改)，这种分层的wheel提供了一个很好的应对并发的结构，比如秒针只会在第一个wheel上获取锁，而在不同wheel上添加任务的线程可以并发执行，每个wheel的锁独立，这种细粒度的锁就和mysql的锁一样，提供粗粒度到细粒度的锁，实际上，在后面的wheel可以提供更加细粒的的slot锁，但是，这个是需要考虑一个问题的，就是，在iwheel上添加任务是不能让当前的指针移动的，这样对tick来说，一旦发生进位，就要锁住整个wheel，移动指针，移动任务，此时的slot也应该是锁上的，所以，依然会有竞争态，但是相比没有slot锁，竞争会小很多，并发度会高很多(但没有实现，后面有机会再实现吧)。

1.1.2 slot迁移

当后面的指针移动时，需要将指向的slot的task移动到前一个wheel，这里实际上就是遍历列表插入操作，不过，这个有一个快速通道，就是不需要判断任务是否在前一个wheel上，一定在前一个wheel，所以直接通过与操作获得索引，所以效率比较快。但是需要注意一点是，进位可能是连续的进位，所以在迁移任务的时候必须要从后往前迁移。不过，这里在实现的时候感觉这个timewheel结构有点问题，觉得有一个类似wheel control管理每一个wheel会更高效一点，这里是通过逐层递归的方式。

1.2 ticker

enum TimeUint : uint64_t{SECONDS = 1000000,MILLISECOND = 1000,MICROSECONDS = 1,
};class Ticker{
public:Ticker() : m_uiInterval(100), m_eUint(MICROSECONDS),m_uiMicroInterval(m_uiInterval*m_eUint)  {}explicit Ticker(uint64_t interval, TimeUint uint): m_uiInterval(interval), m_eUint(uint),m_uiMicroInterval(m_uiInterval*m_eUint){}uint64_t GetInterval() const {return m_uiMicroInterval; }TimeUint GetUint() const {return m_eUint; }virtual void SetInterval(uint64_t interval) {m_uiInterval = interval;m_uiMicroInterval = m_uiInterval*m_eUint;}virtual void GetUint(TimeUint uint) {m_eUint = uint;m_uiMicroInterval = m_uiInterval*m_eUint;}virtual void tick() = 0;virtual void Stop() = 0;protected:uint64_t m_uiInterval;TimeUint m_eUint;uint64_t m_uiMicroInterval;
};

这个到是没想到一个精准的方案，包括一个间隔和单位。提供tick和stop操作。不过，tick的方式很朴实，要么：

通过this_thread::sleep_for休眠。
通过select定时返回，只要select集合里的东西都是NULL就行。
通过linux的timefd定时，start的时候结束就是UINT64_MAX就行。

查了一些资料，看到的好像是这些方法，似乎没有更加精准的定时方法了。

2.3 workerpool

用来处理定时任务的线程池。这里，为了减少对锁的竞争，workerpool维护的任务队列是这种形式

这里是因为，wheel每个slot是指向一个任务队列，当定时任务到时的时候，直接将指向任务队列的指针放到queue末尾。指针的移动能够减小临界区访问长度，这样就不会出现激烈的锁的竞争。而worker取任务的时候，也是直接取走指针。执行完任务，负责delete操作。

2.4 timer

class Timer{
public:Timer(uint64_t worknum, Ticker* ticker, TimerStruct* tmstruct);~Timer();void RegistTimerStruct(TimerStruct* tmstruct);void AddTask(Timestamp& timestamp, TimeDelayTask& task);// void AddTask(Timestamp& timestamp, TimeDelayTask&& task);void AddTaskToQueue(Timestamp& timestamp, TimeDelayTask& task);void AddTaskToQueue(TimeDelayTaskQueue* ptr);void DeleteTask(Timestamp& timestamp);bool Start();bool Stop();bool Clear();uint64_t GetInterval();Timestamp GetTimestamp() { return m_oStartTime; }private:void Tick();inline bool InRunning();uint64_t m_iWorkersNum;uint64_t m_iInterval;std::atomic<uint64_t> m_atoiCount;std::atomic<bool> m_bIsRunning;Timestamp m_oStartTime;Ticker* m_poTicker;TimerStruct* m_poTimerStruct;WorkerPool* m_poWorkerPool;
//    MutexLock m_oMutex;
};

m_iWorkersNum：worker的工作数量。
uint64_t m_iInterval：步长，每个tick的间隔。
m_atoiCount：每tick一次自增一次，通过这个，可以在这一层对定时任务的时间判断，如果小于定时器时间，可以说定时失败(也可以直接当到队列中，立即执行)，等于的任务直接放到任务队列。大于就添加到wheel上。
m_bIsRunning：是否在运行中。
m_oStartTime：开始时间。
m_poTicker：ticker指针。
m_poTimerStruct：struct指针。
m_poWorkerPool：workerpool指针。