C++如何实现带优先级的时间轮定时任务心跳检测调度优化方案

在C++开发的服务端系统中，定时任务调度和心跳检测是支撑业务稳定运行的核心模块。传统基于排序链表或者最小堆的定时调度方案，在处理大量高频定时任务时，会出现插入、删除操作时间复杂度高的问题，而时间轮结构可以将任务触发的时间复杂度降低到O(1)，结合优先级队列处理同时间槽内的任务优先级排序，能够很好地满足高并发场景下的调度需求。

核心设计思路与任务结构定义

整个方案的核心由三个部分组成：时间轮结构、带优先级的定时任务结构、心跳检测调度管理器。时间轮采用分层设计，分为秒级、分级、小时级三层，减少单个时间槽的任务数量。每个定时任务包含触发时间戳、优先级、回调函数三个核心属性。调度管理器负责时间轮的推进和任务的触发执行。这种分层设计不仅降低了单层时间轮的槽位压力，还能高效管理长周期任务，使得系统在处理海量定时任务时依然保持极高的吞吐量。

定时任务需要包含优先级信息，优先级数值越小代表优先级越高，同时需要记录任务的唯一标识和触发时间，方便后续的任务取消操作。在C++中，可以利用函数对象来封装回调逻辑，使得任务触发时能够执行任意指定的业务代码。通过重载比较运算符，可以确保任务在优先级队列中按照时间和优先级双重维度进行排序，保证高优先级任务能够优先被调度执行。

#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个槽，每个槽对应一个优先级队列，存储该时间槽需要触发的任务。这种设计借鉴了钟表的运行原理，通过层级递进的方式管理不同延迟时长的任务。秒轮负责处理短期内的任务触发，而分轮和小时轮则用于管理中长期任务，当低层级时间轮走完一圈时，高层级时间轮便会向下推进，并将对应槽位中的任务降级分配到下一层。

在添加任务时，系统会计算任务触发时间与当前时间的差值。根据延迟的时长，将任务分配到不同层级的时间轮槽位中。如果延迟时间超过了时间轮的最大表示范围，则会被放入最高层级的最后一个槽位中，以防止溢出。这种分配策略确保了任务插入操作的时间复杂度始终保持在一个极低的常数级别，极大地提升了调度性能。

时间轮的推进逻辑通过定时调用tick函数实现。每秒推进一次秒轮，当秒轮走完一圈，推进分轮并将分轮当前槽位的任务重新分配到秒轮中；同理，分轮走完一圈则推进小时轮。在处理当前槽位任务时，会检查任务是否真正到达触发时间，若未到达则重新添加，以处理可能存在的时间偏差，保证任务触发的精准度。

#include <queue>
#include <vector>
#include <chrono>

// 定义三层时间轮的槽数量
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;

    // 获取当前时间戳（毫秒）
    int64_t get_now_ms() {
        return std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()
        ).count();
    }

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 = get_now_ms();
        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 = get_now_ms();
            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::thread tick_thread;
    std::atomic<bool> running;

    int64_t get_now_ms() {
        return std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch()
        ).count();
    }

    // 发送心跳包并注册下一次心跳任务
    void send_heartbeat(uint64_t client_id) {
        std::cout << "Send heartbeat to client: " << client_id << std::endl;
        if (client_states.count(client_id)) {
            int64_t now = get_now_ms();
            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);
        }
    }

    // 检查心跳超时的客户端
    void check_heartbeat_timeout() {
        int64_t now = get_now_ms();
        for (auto it = client_states.begin(); it != client_states.end(); ) {
            if (now - it->second.last_heartbeat_time > it->second.heartbeat_interval * 3) {
                std::cout << "Client timeout: " << it->second.client_id << std::endl;
                it = client_states.erase(it);
            } else {
                ++it;
            }
        }
    }

    // 时间轮推进线程函数
    void tick_loop() {
        while (running) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
            time_wheel.tick();
            check_heartbeat_timeout();
        }
    }

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 = get_now_ms();
        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);
    }

    // 更新客户端心跳时间
    void update_client_heartbeat(uint64_t client_id) {
        if (client_states.count(client_id)) {
            client_states[client_id].last_heartbeat_time = get_now_ms();
        }
    }
};

调度方案优化与实际应用示例

在实际使用中，可以根据场景需求对方案进行进一步优化。例如任务去重优化，如果同一个客户端重复添加心跳任务，可以先取消之前的任务再添加新的，避免重复发送心跳，减少无效任务占用时间槽资源。时间轮槽位动态调整也是一项重要优化，如果业务场景的定时任务集中在某个时间范围，可以动态调整对应层的槽位数量，减少内存占用并提升处理效率。

任务执行线程池化能够显著提升调度吞吐量。时间轮触发任务后，任务执行线程池化能够显著提升调度吞吐量。时间轮触发任务后，不应在时间轮推进线程中直接执行用户回调。时间轮推进通常由单一线程负责，如果某个回调执行时间过长，会阻塞后续槽位的处理，导致定时精度下降。推荐将回调封装为轻量级任务对象，投递到线程池队列中异步执行。这样时间轮线程只负责扫描槽位和分发任务，真正耗时的业务逻辑由工作线程承担。

void TimerWheel::invoke_task(const TimerTask& task) {
    if (task.priority == TaskPriority::HIGH) {
        high_priority_executor->submit([task]() {
            task.callback();
        });
    } else {
        normal_executor->submit([task]() {
            task.callback();
        });
    }
}

线程池参数同样需要根据业务特点调优。IO 密集型任务可以适当增加线程数，避免任务在队列中长时间等待；CPU 密集型任务则不宜超过可用核心数太多，否则上下文切换开销会抵消并发收益。线程池队列应设置合理上限，当任务堆积达到阈值时，可以采用拒绝策略或降级处理，例如丢弃低优先级任务、合并重复任务或暂时关闭非关键定时器，防止内存持续增长。 实际应用中，时间轮除了用于心跳检测，还广泛用于连接超时检测、报文重传、订单超时关闭和游戏技能冷却等场景。连接超时检测与心跳检测类似，但通常需要在收到数据包后重置定时任务，实现动态超时管理。下面是一个基于时间轮的连接超时检测示例，展示了任务去重和定时器重置的实际用法。

void ConnectionManager::add_connection(uint64_t conn_id) {
    uint64_t now = get_now_ms();
    client_states[conn_id] = {conn_id, now, timeout_ms};

    TimerTask task;
    task.task_id = task_id_generator++;
    task.trigger_time = now + timeout_ms;
    task.priority = TaskPriority::NORMAL;
    task.callback = [this, conn_id]() { check_connection_timeout(conn_id); };

    client_states[conn_id].last_timeout_task = task.task_id;
    time_wheel.add_task(task);
}

void ConnectionManager::on_packet_received(uint64_t conn_id) {
    auto it = client_states.find(conn_id);
    if (it == client_states.end()) return;

    uint64_t now = get_now_ms();
    it->second.last_active_time = now;

    // 取消旧任务并添加新任务，避免重复触发
    time_wheel.cancel_task(it->second.last_timeout_task);

    TimerTask task;
    task.task_id = task_id_generator++;
    task.trigger_time = now + timeout_ms;
    task.priority = TaskPriority::NORMAL;
    task.callback = [this, conn_id]() { check_connection_timeout(conn_id); };

    it->second.last_timeout_task = task.task_id;
    time_wheel.add_task(task);
}

void ConnectionManager::check_connection_timeout(uint64_t conn_id) {
    auto it = client_states.find(conn_id);
    if (it == client_states.end()) return;

    uint64_t now = get_now_ms();
    if (now - it->second.last_active_time >= timeout_ms) {
        close_connection(conn_id);
        client_states.erase(it);
    }
}

通过以上设计，时间轮调度器可以将大量定时任务均匀分散到多层时间槽中，保证插入、删除和触发操作的时间复杂度接近常数级。对于百万级并发连接和大量短周期定时任务，时间轮相比传统红黑树或最小堆实现，能够显著降低定时器管理的 CPU 开销，同时避免频繁重建堆结构带来的性能抖动。 在实际工程中，时间轮方案通常还需要与事件循环协同工作。例如在 Linux 平台上，可以将时间轮最近一次到期时间转换为 epoll 的超时参数，使得事件循环在没有网络事件时也能准时唤醒并推进时间轮。这样定时触发和网络 IO 复用同一条线程，可以减少锁竞争和线程切换，提高单机资源利用效率。 总的来说，时间轮是一种高效且可扩展的定时任务调度方案，特别适合心跳检测、连接超时、重传定时器等高频短周期场景。实现时需要合理设计层数、槽位粒度和任务优先级，同时配合任务去重、动态槽位调整和线程池化执行，才能在保证定时精度的同时获得较高的吞吐量。不同业务场景下，还应结合事件循环、线程模型和资源限制进行整体权衡，最终形成稳定、可维护的定时调度模块。

C++时间轮定时任务调度心跳检测优先级队列修改时间：2026-07-19 04:39:57