Linux 高精准定时任务线程池设计详解epolltimerfd小根堆一、前言定时线程池要解决什么问题普通线程池只能立即执行任务无法实现延迟 N 毫秒执行任务按时间序调度大量任务高精度、低 CPU 占用的定时触发常见开源方案muduo、nginx、redis的定时调度核心思想一致小根堆 事件驱动定时器 独立调度线程 工作线程池本文讲解我们最终实现的Linux 专属高精度定时线程池整套设计思想与底层原理。二、核心基础概念必须掌握1. 定时任务的本质所有定时任务只有一个核心属性到期时间点 due_time。谁的 due_time 更小谁先执行。因此定时任务天然适配小根堆。2. 为什么必须用小根堆堆顶永远是最近要到期的任务插入任务 O(logN)取出最近任务 O(1)完全符合定时调度场景。3. Linux 三大事件驱动组件作用1timerfd高精度定时器时间事件将时间流逝转为文件可读事件支持纳秒级定时。优点不占用线程、不自旋、内核休眠、时间精准。2eventfd手动唤醒信号外部事件用于新增任务、停止队列时主动唤醒调度线程。解决核心问题新任务比当前堆顶更早到期时需要立刻刷新定时器。3epoll事件多路统一监听同时监听timerfd时间到 eventfd新任务/停止。实现单线程无轮询、零CPU空闲占用。三、整体架构分层核心设计精髓整套架构分为三层解耦工业级标准设计1. 任务层双队列模型延迟堆 wait_delay_q_保存未到期任务小根堆就绪队列 ready_run_q_保存已到期、可立刻执行任务FIFO目的定时迁移与任务执行彻底解耦。2. 调度层epoll 事件调度线程1 条只做一件事检测时间到期迁移任务到就绪队列。不执行业务、不阻塞、轻量、高精度。3. 执行层工作线程池N 条不断从就绪队列取任务、执行业务逻辑。四、核心设计逻辑详解1. 为什么采用「双队列」如果只用一个堆工作线程取任务需要频繁加锁堆堆结构复杂、并发性能差到期任务批量弹出时会阻塞调度双队列优势调度线程只操作堆写多、读少工作线程只操作普通队列高并发消费互不干扰并发性能极高2. 为什么必须 epoll timerfd不能 whilesleep很多初学者写定时任务用死循环 sleep 扫描固定 10ms 扫描 → 精度极低为了高精度必须 sleep(1ms) → CPU 跑满新任务无法即时响应epolltimerfd 方案无事件时线程完全休眠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. 停止逻辑必须优雅唤醒所有阻塞线程否则线程永久阻塞、程序无法退出、内存泄漏。七、各方案优劣横向对比方案精度CPU占用动态任务响应工业级可用性whilesleep 轮询低极高差不可用条件变量 wait_for中中一般一般epolltimerfd小根堆极高(ns)极低(空闲0%)实时响应工业级八、最终架构设计总结基于 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::functionvoid(); 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_queueTimerTask, std::vectorTimerTask, TimerCmp wait_delay_q_; std::queueTimerTask 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::atomicbool 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_castMs(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_lockstd::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_lockstd::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_castTimerTask(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_lockstd::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_lockstd::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_lockstd::mutex lock(mtx_); cv_empty_.wait(lock, [this](){ return !ready_run_q_.empty() || !running_; }); if (!ready_run_q_.empty()) { TimerTask tmp std::move(const_castTimerTask(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::functionvoid(); 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::atomicbool running_; std::vectorstd::thread workers_; }; #endif