基于C11的线程池模型 - 计划式线程池

Linux 高精准定时任务线程池设计详解（epoll+timerfd+小根堆）

一、前言：定时线程池要解决什么问题

普通线程池只能立即执行任务，无法实现：

• 延迟 N 毫秒执行任务

• 按时间序调度大量任务

• 高精度、低 CPU 占用的定时触发

常见开源方案（muduo、nginx、redis）的定时调度核心思想一致：

小根堆 + 事件驱动定时器 + 独立调度线程 + 工作线程池

本文讲解我们最终实现的Linux 专属高精度定时线程池整套设计思想与底层原理。

二、核心基础概念（必须掌握）

1. 定时任务的本质

所有定时任务只有一个核心属性：到期时间点 due_time。

谁的 due_time 更小，谁先执行。

因此定时任务天然适配小根堆。

2. 为什么必须用小根堆？

• 堆顶永远是最近要到期的任务

• 插入任务 O(logN)

• 取出最近任务 O(1)

完全符合定时调度场景。

3. Linux 三大事件驱动组件作用

（1）timerfd：高精度定时器（时间事件）

将时间流逝转为文件可读事件，支持纳秒级定时。

优点：不占用线程、不自旋、内核休眠、时间精准。

（2）eventfd：手动唤醒信号（外部事件）

用于：新增任务、停止队列时，主动唤醒调度线程。

解决核心问题：新任务比当前堆顶更早到期时，需要立刻刷新定时器。

（3）epoll：事件多路统一监听

同时监听：timerfd（时间到）+ eventfd（新任务/停止）。

实现：单线程无轮询、零CPU空闲占用。

三、整体架构分层（核心设计精髓）

整套架构分为三层解耦（工业级标准设计）：

1. 任务层：双队列模型

• 延迟堆 wait_delay_q_：保存未到期任务（小根堆）

• 就绪队列 ready_run_q_：保存已到期、可立刻执行任务（FIFO）

目的：定时迁移与任务执行彻底解耦。

2. 调度层：epoll 事件调度线程（1 条）

只做一件事：检测时间到期，迁移任务到就绪队列。

不执行业务、不阻塞、轻量、高精度。

3. 执行层：工作线程池（N 条）

不断从就绪队列取任务、执行业务逻辑。

四、核心设计逻辑详解

1. 为什么采用「双队列」？

如果只用一个堆：

• 工作线程取任务需要频繁加锁堆

• 堆结构复杂、并发性能差

• 到期任务批量弹出时会阻塞调度

双队列优势：

• 调度线程只操作堆（写多、读少）

• 工作线程只操作普通队列（高并发消费）

• 互不干扰，并发性能极高

2. 为什么必须 epoll + timerfd，不能 while+sleep？

很多初学者写定时任务用死循环 sleep 扫描：

• 固定 10ms 扫描 → 精度极低

• 为了高精度必须 sleep(1ms) → CPU 跑满

• 新任务无法即时响应

epoll+timerfd 方案：

• 无事件时线程完全休眠，CPU 0%

• 时间精度纳秒级

• 新增更早任务可立刻唤醒刷新定时器

3. 为什么必须 eventfd？

假设当前堆顶任务 5 秒后到期，timerfd 阻塞 5 秒。

此时用户新增一个 1 秒后执行的任务：

如果没有 eventfd 唤醒：

调度线程继续阻塞 5 秒 → 1 秒任务被严重延迟！

eventfd 的唯一核心作用：打断当前阻塞，重新刷新最近定时时间。

4. 小根堆排序逻辑

排序依据：绝对到期时间 due_time

规则：越早到期，优先级越高。

每次更新 timerfd 只需要看：堆顶.due_time

五、完整运行流程（全链路复盘）

步骤1：用户提交延迟任务

任务计算 due_time，存入延迟小根堆。

eventfd 唤醒调度线程。

步骤2：调度线程刷新定时器

取堆顶最近到期时间，重置 timerfd 超时时间。

epoll 再次休眠等待事件。

步骤3：时间到达，timerfd 触发

epoll 唤醒调度线程。

批量遍历堆，把所有过期任务迁移到就绪队列。

唤醒所有工作线程。

步骤4：工作线程消费任务

多线程并发从就绪队列取任务执行。

六、关键难点与工程坑点（面试高频）

1. timerfd 触发后必须 read

不 read 清空计数器，fd 会一直可读 → epoll 死循环触发。

2. priority_queue top() 返回 const 引用

无法直接 move，必须安全 const_cast 转移资源，否则编译报错。

3. 虚假唤醒问题

所有 condition_variable 等待必须带 lambda 谓词判断，不能裸等待。

4. 新增更早任务必须刷新定时器

这是 eventfd 存在的唯一意义，也是普通定时器方案最大漏洞。

5. 停止逻辑必须优雅唤醒所有阻塞线程

否则线程永久阻塞、程序无法退出、内存泄漏。

七、各方案优劣横向对比

八、最终架构设计总结

基于 Linux 原生事件机制，设计了一套高精度、低功耗、线程安全、可动态调度的定时任务线程池：

• 采用小根堆管理延迟任务，保证最近任务优先触发

• 使用timerfd实现高精度内核定时器，无空轮询

• 使用eventfd实现动态刷新定时时间，支持随时新增优先任务

• 通过epoll统一监听时间事件与唤醒事件，单调度线程休眠等待

• 采用延迟堆+就绪队列双队列解耦，分离调度与执行职责

• 结合生产者消费者条件变量模型，实现线程安全的高并发任务提交与消费

整体架构满足：高精度、低CPU、高并发、可扩展、无泄漏、优雅退出。

附上源码

#ifndef _SYNC_TIMER_QUEUE_HPP_ #define _SYNC_TIMER_QUEUE_HPP_ #ifdef __linux__ #include <sys/epoll.h> #include <sys/eventfd.h> #include <sys/timerfd.h> #include <unistd.h> #include <cerrno> #endif #include <atomic> #include <condition_variable> #include <functional> #include <mutex> #include <queue> #include <thread> #include <vector> #include <chrono> #include <utility> #include <cstdint> using TaskFunc = std::function<void()>; using Clock = std::chrono::steady_clock; using TimePoint = Clock::time_point; using Ms = std::chrono::milliseconds; // 定时任务结构体 struct TimerTask { uint64_t delay_ms; TaskFunc task; TimePoint due_time; explicit TimerTask(uint64_t delay, TaskFunc&& func) : delay_ms(delay), task(std::move(func)), due_time(Clock::now() + Ms(delay)) {} TimerTask() : delay_ms(0), task(nullptr), due_time(Clock::now()) {} TimerTask(const TimerTask&) = default; TimerTask& operator=(const TimerTask&) = default; TimerTask(TimerTask&&) noexcept = default; TimerTask& operator=(TimerTask&&) noexcept = default; }; // 小根堆比较：到期越早优先级越高 struct TimerCmp { bool operator()(const TimerTask& lhs, const TimerTask& rhs) const noexcept { return lhs.due_time > rhs.due_time; } }; class SyncTimerQueue { private: // 双队列：延迟堆 + 就绪执行队列 std::priority_queue<TimerTask, std::vector<TimerTask>, TimerCmp> wait_delay_q_; std::queue<TimerTask> ready_run_q_; const size_t max_cap_; const size_t enqueue_wait_ms_; mutable std::mutex mtx_; std::condition_variable cv_full_; std::condition_variable cv_empty_; std::atomic<bool> running_{true}; #ifdef __linux__ int epoll_fd_ = -1; int timer_fd_ = -1; int wake_efd_ = -1; std::thread schedule_th_; #endif bool IsFull() const { return wait_delay_q_.size() + ready_run_q_.size() >= max_cap_; } bool IsAllEmpty() const { return wait_delay_q_.empty() && ready_run_q_.empty(); } #ifdef __linux__ // 更新timerfd为最近到期任务 bool ResetTimerFd() { if (wait_delay_q_.empty()) { itimerspec zero{}; return timerfd_settime(timer_fd_, 0, &zero, nullptr) == 0; } auto now = Clock::now(); const auto& top_task = wait_delay_q_.top(); itimerspec spec{}; if (top_task.due_time <= now) { spec.it_value.tv_nsec = 1; } else { auto diff = std::chrono::duration_cast<Ms>(top_task.due_time - now).count(); spec.it_value.tv_sec = diff / 1000; spec.it_value.tv_nsec = (diff % 1000) * 1000000; } return timerfd_settime(timer_fd_, 0, &spec, nullptr) == 0; } // 唤醒调度线程 void WakeScheduler() { uint64_t sig = 1; (void)write(wake_efd_, &sig, sizeof(sig)); } // 调度主循环 epoll 事件驱动 void ScheduleLoop() { constexpr int MAX_EV = 2; epoll_event evs[MAX_EV]; while (true) { std::unique_lock<std::mutex> lock(mtx_); if (!running_ && IsAllEmpty()) break; if (!ready_run_q_.empty()) cv_empty_.notify_one(); if (!running_ && ready_run_q_.empty()) break; ResetTimerFd(); lock.unlock(); // 500ms兜底超时，防止卡死 int n = epoll_wait(epoll_fd_, evs, MAX_EV, 500); if (n <= 0) continue; std::unique_lock<std::mutex> elock(mtx_); for (int i = 0; i < n; ++i) { int fd = evs[i].data.fd; if (fd == timer_fd_) { // 定时器到期，迁移所有过期任务 uint64_t cnt = 0; (void)read(timer_fd_, &cnt, sizeof(cnt)); auto now = Clock::now(); while (!wait_delay_q_.empty() && wait_delay_q_.top().due_time <= now) { TimerTask move_t = std::move(const_cast<TimerTask&>(wait_delay_q_.top())); wait_delay_q_.pop(); ready_run_q_.push(std::move(move_t)); } cv_empty_.notify_all(); } else if (fd == wake_efd_) { uint64_t val = 0; (void)read(wake_efd_, &val, sizeof(val)); } } } } #endif // 底层入队封装 int InnerEnqueue(TimerTask&& task) { std::unique_lock<std::mutex> lock(mtx_); bool ok = cv_full_.wait_for(lock, Ms(enqueue_wait_ms_), [this](){ return !IsFull() || !running_; }); if (!ok) return 1; if (!running_) return 2; wait_delay_q_.push(std::move(task)); lock.unlock(); #ifdef __linux__ WakeScheduler(); #endif cv_empty_.notify_one(); return 0; } public: SyncTimerQueue(size_t max_cap, size_t enqueue_ms = 1000) : max_cap_(max_cap), enqueue_wait_ms_(enqueue_ms) { #ifdef __linux__ epoll_fd_ = epoll_create1(EPOLL_CLOEXEC); timer_fd_ = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); wake_efd_ = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); if (epoll_fd_ >= 0 && timer_fd_ >= 0 && wake_efd_ >= 0) { epoll_event ev{}; ev.events = EPOLLIN; ev.data.fd = timer_fd_; epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &ev); ev.data.fd = wake_efd_; epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, wake_efd_, &ev); schedule_th_ = std::thread(&SyncTimerQueue::ScheduleLoop, this); } #endif } ~SyncTimerQueue() { Stop(); #ifdef __linux__ if (schedule_th_.joinable()) schedule_th_.join(); if (epoll_fd_ >= 0) close(epoll_fd_); if (timer_fd_ >= 0) close(timer_fd_); if (wake_efd_ >= 0) close(wake_efd_); #endif } // 停止队列，唤醒所有阻塞线程 void Stop() { { std::unique_lock<std::mutex> lock(mtx_); running_ = false; } #ifdef __linux__ WakeScheduler(); #endif cv_full_.notify_all(); cv_empty_.notify_all(); } // 对外提交任务接口 int PutDelayTask(const TimerTask& task) { return InnerEnqueue(TimerTask(task)); } int PutDelayTask(TimerTask&& task) { return InnerEnqueue(std::move(task)); } int PutDelayTask(uint64_t delay_ms, TaskFunc&& func) { return PutDelayTask(TimerTask(delay_ms, std::move(func))); } // 阻塞获取就绪任务 int TakeTask(TimerTask& out) { std::unique_lock<std::mutex> lock(mtx_); cv_empty_.wait(lock, [this](){ return !ready_run_q_.empty() || !running_; }); if (!ready_run_q_.empty()) { TimerTask tmp = std::move(const_cast<TimerTask&>(ready_run_q_.front())); ready_run_q_.pop(); out = std::move(tmp); cv_full_.notify_one(); return 0; } return 2; } SyncTimerQueue(const SyncTimerQueue&) = delete; SyncTimerQueue& operator=(const SyncTimerQueue&) = delete; SyncTimerQueue(SyncTimerQueue&&) = delete; SyncTimerQueue& operator=(SyncTimerQueue&&) = delete; }; #endif

#ifndef _SCHEDULE_THREAD_POOL_HPP_ #define _SCHEDULE_THREAD_POOL_HPP_ #include "SyncQueue4.hpp" #include <atomic> #include <vector> #include <thread> #include <functional> using TaskFunc = std::function<void()>; class ScheduleThreadPool { public: /// @brief 构造线程池 /// @param worker_num 工作线程数量 /// @param max_task 队列最大容纳任务总数 /// @param enqueue_ms 入队等待超时(ms) explicit ScheduleThreadPool(size_t worker_num, size_t max_task, size_t enqueue_ms = 1000) : queue_(max_task, enqueue_ms), running_(true) { workers_.reserve(worker_num); for (size_t i = 0; i < worker_num; ++i) { workers_.emplace_back(&ScheduleThreadPool::WorkerLoop, this); } } // 禁止拷贝移动 ScheduleThreadPool(const ScheduleThreadPool&) = delete; ScheduleThreadPool& operator=(const ScheduleThreadPool&) = delete; ScheduleThreadPool(ScheduleThreadPool&&) = delete; ScheduleThreadPool& operator=(ScheduleThreadPool&&) = delete; ~ScheduleThreadPool() { Stop(); } /// 提交延迟任务 /// @param delay_ms 延迟毫秒 /// @param func 待执行函数 /// @return 0成功 1入队超时 2队列已停止 int SubmitDelayTask(uint64_t delay_ms, TaskFunc&& func) { return queue_.PutDelayTask(delay_ms, std::move(func)); } /// 停止线程池与定时队列 void Stop() { if (!running_.exchange(false)) return; queue_.Stop(); // 回收所有工作线程 for (auto& th : workers_) { if (th.joinable()) th.join(); } } private: // 工作线程循环：阻塞从队列取就绪任务执行 void WorkerLoop() { TimerTask task; while (running_) { int ret = queue_.TakeTask(task); if (ret != 0) break; // 执行业务任务 if (task.task) task.task(); } } private: SyncTimerQueue queue_; std::atomic<bool> running_; std::vector<std::thread> workers_; }; #endif