在C++开发的服务端系统中，定时任务调度和心跳检测是支撑业务稳定运行的核心模块。传统基于排序链表或者最小堆的定时调度方案，在处理大量高频定时任务时，会出现插入、删除操作时间复杂度高的问题，而时间轮结构可以将任务触发的时间复杂度降低到O(1)，结合优先级队列处理同时间槽内的任务优先级排序，能够很好地满足高并发场景下的调度需求。
核心设计思路
整个方案的核心由三个部分组成：时间轮结构、带优先级的定时任务结构、心跳检测调度管理器。时间轮采用分层设计，分为秒级、分级、小时级三层，减少单个时间槽的任务数量；每个定时任务包含触发时间戳、优先级、回调函数三个核心属性；调度管理器负责时间轮的推进和任务的触发执行。
任务结构设计
定时任务需要包含优先级信息，优先级数值越小代表优先级越高，同时需要记录任务的唯一标识和触发时间，方便后续的任务取消操作。
#include <functional>
#include <chrono>
// 定时任务优先级枚举，数值越小优先级越高
enum class TaskPriority {
HIGH = 0,
NORMAL = 1,
LOW = 2
};
// 定时任务结构
struct TimerTask {
uint64_t task_id; // 任务唯一ID
int64_t trigger_time; // 触发时间戳（毫秒）
TaskPriority priority; // 任务优先级
std::function<void()> callback; // 任务回调函数
// 优先级比较函数，用于优先级队列排序
bool operator<(const TimerTask& other) const {
// 先按触发时间排序，触发时间相同则按优先级排序
if (trigger_time != other.trigger_time) {
return trigger_time > other.trigger_time;
}
return static_cast<int>(priority) > static_cast<int>(other.priority);
}
};
时间轮结构设计
采用三层时间轮设计，秒轮有60个槽，分轮有60个槽，小时轮有24个槽，每个槽对应一个优先级队列，存储该时间槽需要触发的任务。
#include <queue>
#include <vector>
#include <unordered_map>
// 时间轮三层槽数量定义
const int SECOND_SLOT_NUM = 60;
const int MINUTE_SLOT_NUM = 60;
const int HOUR_SLOT_NUM = 24;
// 时间轮结构
class TimeWheel {
private:
// 三层时间轮的槽，每个槽是优先级队列
std::vector<std::priority_queue<TimerTask>> second_slots;
std::vector<std::priority_queue<TimerTask>> minute_slots;
std::vector<std::priority_queue<TimerTask>> hour_slots;
int current_second;
int current_minute;
int current_hour;
public:
TimeWheel()
: second_slots(SECOND_SLOT_NUM),
minute_slots(MINUTE_SLOT_NUM),
hour_slots(HOUR_SLOT_NUM),
current_second(0),
current_minute(0),
current_hour(0) {}
// 添加任务到时间轮
void add_task(const TimerTask& task) {
int64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch()
).count();
int64_t delay = task.trigger_time - now;
if (delay < 0) {
// 任务已经过期，直接执行
task.callback();
return;
}
// 计算延迟的时分秒
int delay_sec = delay / 1000;
int delay_hour = delay_sec / 3600;
int delay_minute = (delay_sec % 3600) / 60;
int delay_second = delay_sec % 60;
if (delay_hour >= HOUR_SLOT_NUM) {
// 超过时间轮最大范围，放入小时轮最后一个槽
hour_slots[HOUR_SLOT_NUM - 1].push(task);
} else if (delay_hour > 0) {
int slot = (current_hour + delay_hour) % HOUR_SLOT_NUM;
hour_slots[slot].push(task);
} else if (delay_minute > 0) {
int slot = (current_minute + delay_minute) % MINUTE_SLOT_NUM;
minute_slots[slot].push(task);
} else {
int slot = (current_second + delay_second) % SECOND_SLOT_NUM;
second_slots[slot].push(task);
}
}
// 推进时间轮，每秒调用一次
void tick() {
// 处理当前秒槽的任务
auto& sec_queue = second_slots[current_second];
while (!sec_queue.empty()) {
TimerTask task = sec_queue.top();
sec_queue.pop();
int64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch()
).count();
if (task.trigger_time <= now) {
task.callback();
} else {
// 任务还没到触发时间，重新添加回时间轮
add_task(task);
}
}
current_second = (current_second + 1) % SECOND_SLOT_NUM;
if (current_second == 0) {
// 秒轮转完一圈，推进分轮
current_minute = (current_minute + 1) % MINUTE_SLOT_NUM;
// 将分轮当前槽的任务转移到秒轮
auto& min_queue = minute_slots[current_minute];
while (!min_queue.empty()) {
TimerTask task = min_queue.top();
min_queue.pop();
add_task(task);
}
if (current_minute == 0) {
// 分轮转完一圈，推进小时轮
current_hour = (current_hour + 1) % HOUR_SLOT_NUM;
// 将小时轮当前槽的任务转移到分轮
auto& hour_queue = hour_slots[current_hour];
while (!hour_queue.empty()) {
TimerTask task = hour_queue.top();
hour_queue.pop();
add_task(task);
}
}
}
}
};
心跳检测调度实现
心跳检测通常需要定期向客户端发送心跳包，同时检测客户端的心跳响应是否超时。我们可以把心跳发送任务作为定时任务放入时间轮，同时维护一个客户端心跳状态表，记录每个客户端最后一次收到心跳的时间。
#include <unordered_map>
#include <thread>
#include <atomic>
#include <iostream>
// 客户端心跳状态
struct ClientHeartbeatState {
uint64_t client_id;
int64_t last_heartbeat_time; // 最后一次收到心跳的时间（毫秒）
int heartbeat_interval; // 心跳发送间隔（毫秒）
};
// 心跳检测调度管理器
class HeartbeatScheduler {
private:
TimeWheel time_wheel;
std::unordered_map<uint64_t, ClientHeartbeatState> client_states;
std::atomic<uint64_t> task_id_generator;
std::unordered_map<uint64_t, uint64_t> task_client_map; // 任务ID到客户端ID的映射
std::thread tick_thread;
std::atomic<bool> running;
// 发送心跳包的回调
void send_heartbeat(uint64_t client_id) {
std::cout << "Send heartbeat to client: " << client_id << std::endl;
// 实际场景中这里会调用网络发送接口
// 发送完成后，重新添加下一个心跳任务
int64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch()
).count();
if (client_states.count(client_id)) {
int64_t next_trigger = now + client_states[client_id].heartbeat_interval;
TimerTask task;
task.task_id = task_id_generator++;
task.trigger_time = next_trigger;
task.priority = TaskPriority::NORMAL;
task.callback = [this, client_id]() { send_heartbeat(client_id); };
time_wheel.add_task(task);
task_client_map[task.task_id] = client_id;
}
}
// 时间轮推进线程函数
void tick_loop() {
while (running) {
std::this_thread::sleep_for(std::chrono::seconds(1));
time_wheel.tick();
// 检查心跳超时的客户端
check_heartbeat_timeout();
}
}
// 检查心跳超时的客户端
void check_heartbeat_timeout() {
int64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch()
).count();
for (auto it = client_states.begin(); it != client_states.end(); ) {
// 超过3个心跳周期没收到响应则判定超时
if (now - it->second.last_heartbeat_time > it->second.heartbeat_interval * 3) {
std::cout << "Client timeout: " << it->second.client_id << std::endl;
// 移除该客户端的所有任务
remove_client_tasks(it->second.client_id);
it = client_states.erase(it);
} else {
++it;
}
}
}
// 移除客户端相关的所有任务
void remove_client_tasks(uint64_t client_id) {
// 实际场景中需要维护任务列表，这里简化逻辑
// 可以在TimerTask中添加client_id字段，时间轮推进时过滤已移除的客户端任务
}
public:
HeartbeatScheduler() : task_id_generator(0), running(false) {}
// 启动调度器
void start() {
running = true;
tick_thread = std::thread(&HeartbeatScheduler::tick_loop, this);
}
// 停止调度器
void stop() {
running = false;
if (tick_thread.joinable()) {
tick_thread.join();
}
}
// 添加客户端心跳检测
void add_client(uint64_t client_id, int heartbeat_interval) {
int64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch()
).count();
client_states[client_id] = {client_id, now, heartbeat_interval};
// 添加第一个心跳任务
TimerTask task;
task.task_id = task_id_generator++;
task.trigger_time = now + heartbeat_interval;
task.priority = TaskPriority::NORMAL;
task.callback = [this, client_id]() { send_heartbeat(client_id); };
time_wheel.add_task(task);
task_client_map[task.task_id] = client_id;
}
// 更新客户端心跳时间
void update_client_heartbeat(uint64_t client_id) {
if (client_states.count(client_id)) {
int64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch()
).count();
client_states[client_id].last_heartbeat_time = now;
}
}
};
优化调优方案
在实际使用中，可以根据场景需求对方案进行进一步优化：
- 任务去重优化：如果同一个客户端重复添加心跳任务，可以先取消之前的任务再添加新的，避免重复发送心跳，减少无效任务占用时间槽资源。
- 时间轮槽位动态调整：如果业务场景的定时任务集中在某个时间范围，可以动态调整对应层的槽位数量，减少内存占用。
- 任务执行线程池化：时间轮触发任务后，将任务放入线程池执行，避免任务执行时间过长阻塞时间轮的推进，提升调度吞吐量。
- 优先级队列优化：如果同时间槽内的任务数量非常多，可以使用更高效的优先级队列实现，比如斐波那契堆，降低插入和删除的时间复杂度。
使用示例
以下是调度器的简单使用示例，模拟两个客户端的心跳检测场景：
int main() {
HeartbeatScheduler scheduler;
scheduler.start();
// 添加两个客户端，心跳间隔分别为1秒和2秒
scheduler.add_client(1001, 1000);
scheduler.add_client(1002, 2000);
// 模拟客户端1在3秒后上报一次心跳
std::this_thread::sleep_for(std::chrono::seconds(3));
scheduler.update_client_heartbeat(1001);
// 运行10秒后停止
std::this_thread::sleep_for(std::chrono::seconds(10));
scheduler.stop();
return 0;
}