基于C++20标准实现Linux环境下任务调度功能的方案咨询

核心设计思路 #

• 用最小堆（优先级队列）存储待执行任务，按任务执行时间从小到大排序，保证每次取到的都是最快要执行的任务
• 后台常驻1个调度线程，通过条件变量等待直到最近任务的执行时间，到期后触发任务执行
• 支持提交任意可调用对象+参数，自动打包为无参任务函数存储
• 任务触发后由独立线程运行，不会阻塞调度线程的后续计时
• 基于C++20 std::jthread、std::stop_token实现线程的安全启停，无需手动管理join/detach

调度器完整实现代码 #

#include <chrono>
#include <functional>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <string>
#include <ctime>

class TaskScheduler {
private:
    // 任务结构体
    struct ScheduledTask {
        std::chrono::system_clock::time_point execute_time;
        std::function<void()> task;

        // 重载比较运算符，用于优先级队列最小堆排序
        bool operator<(const ScheduledTask& other) const {
            return execute_time > other.execute_time; // 优先级队列默认是最大堆，反过来实现最小堆
        }
    };

    std::priority_queue<ScheduledTask> m_task_queue;
    std::mutex m_queue_mutex;
    std::condition_variable m_cv;
    std::jthread m_worker_thread;

    // 调度线程工作函数
    void worker_loop(std::stop_token stop_token) {
        while (!stop_token.stop_requested()) {
            std::unique_lock lock(m_queue_mutex);
            // 队列空就一直等新任务
            if (m_task_queue.empty()) {
                m_cv.wait(lock, [this, &stop_token](){
                    return stop_token.stop_requested() || !m_task_queue.empty();
                });
                continue;
            }

            // 取最近要执行的任务
            const auto next_task_time = m_task_queue.top().execute_time;
            // 等待到任务执行时间或者有新任务插入/停止信号
            if (m_cv.wait_until(lock, next_task_time, [this, &stop_token, next_task_time](){
                return stop_token.stop_requested() || 
                       (!m_task_queue.empty() && m_task_queue.top().execute_time != next_task_time);
            })) {
                // 被新任务或停止信号唤醒，重新循环判断
                continue;
            }

            // 到执行时间了，取出任务
            auto task = std::move(m_task_queue.top().task);
            m_task_queue.pop();
            lock.unlock();

            // 独立线程运行任务，不阻塞调度线程
            std::jthread task_thread(std::move(task));
            task_thread.detach(); // detach让任务独立运行，也可替换为线程池管理减少开销
        }
    }

public:
    TaskScheduler() {
        m_worker_thread = std::jthread(&TaskScheduler::worker_loop, this);
    }

    ~TaskScheduler() {
        // 停止调度线程，自动join
        m_worker_thread.request_stop();
        m_cv.notify_all();
    }

    // 对外提交任务接口：支持任意可调用对象+参数 + unix epoch秒级时间戳
    template<typename Func, typename... Args>
    void schedule(Func&& func, Args&&... args, std::int64_t unix_epoch_seconds) {
        // 打包可调用对象和参数为无参函数
        auto bound_task = std::bind_front(std::forward<Func>(func), std::forward<Args>(args)...);
        ScheduledTask task{
            .execute_time = std::chrono::system_clock::from_time_t(unix_epoch_seconds),
            .task = std::move(bound_task)
        };

        {
            std::lock_guard lock(m_queue_mutex);
            m_task_queue.push(std::move(task));
        }
        // 唤醒调度线程，检查是否有更早的任务
        m_cv.notify_one();
    }
};

现有主循环集成示例 #

你原有的主循环代码只需要初始化调度器实例，然后在需要提交任务的位置调用schedule接口即可：

#include <iostream>
#include <vector>
// 假设你自己的Misc工具类头文件
#include "misc.h"

int main() {
    // 全局初始化调度器，生命周期和主程序一致
    TaskScheduler scheduler;

    size_t checked = 0;
    std::vector<std::string> sublist; // 你的任务列表，按原有逻辑填充

    while (sublist.size() - 1 > checked) {
        std::vector<std::string> u_time_split = Misc::split_string(sublist[checked]);

        std::int64_t current_epoch = std::time(nullptr);
        std::int64_t task_run_epoch = std::stoll(u_time_split[1]);

        if (task_run_epoch - current_epoch < 2) {
            std::cout << "Scheduling Thread to run using this: " << u_time_split[0] << " | When to run: " << Misc::epoch_to_utc(u_time_split[1]) << std::endl;

            // 提交任务示例：替换为你实际要执行的任务逻辑即可
            scheduler.schedule([task_id = u_time_split[0]]() {
                // 此处编写任务逻辑，会在指定时间点由独立线程执行
                std::cout << "Executing task: " << task_id << std::endl;
            }, task_run_epoch);
        }

        checked++;
        std::this_thread::sleep_for(std::chrono::seconds(5));
    }

    return 0;
}

注意事项 #

• 如果提交的任务时间戳早于当前时间，任务会被立即触发执行
• 示例中任务线程用了detach，如果你需要跟踪任务执行状态、获取返回值，可以替换为线程池实现，避免频繁创建销毁线程的开销
• 如果任务执行抛出异常，需要在任务函数内部自行捕获处理，避免程序崩溃
• C++20的std::bind_front也可以替换为lambda表达式实现参数绑定，兼容性更好

内容的提问来源于stack exchange，提问作者CoN