定时器中主要的数据结构
- 优先级任务队列:队列中存储任务,每个任务会添加时间戳,最近的时间戳的任务会先出队。
- 锁和条件变量:当有任务需要执行时,用于通知正在等待的线程从任务队列中取出任务执行。
- 线程池:各个任务会放在线程池中执行。
下面是相关代码:
class TimerQueue {public:struct InternalS {std::chrono::time_point<std::chrono::high_resolution_clock> time_point_;std::function<void()> func_;bool operator<(const InternalS& b) const { return time_point_ > b.time_point_; }};enum class RepeatedIdState { kInit = 0, kRunning = 1, kStop = 2 };private:std::priority_queue<InternalS> queue_;bool running_ = false;std::mutex mutex_;std::condition_variable cond_;wzq::ThreadPool thread_pool_;std::atomic<int> repeated_func_id_;wzq::ThreadSafeMap<int, RepeatedIdState> repeated_id_state_map_;};
初始化
在构造函数中初始化,主要是配置好内部的线程池,线程池中常驻的线程数目前设为4。
TimerQueue() : running_(true), thread_pool_(wzq::ThreadPool::ThreadPoolConfig{4, 4, 40, std::chrono::seconds(4)}) {repeated_func_id_.store(0);}
开启定时器功能
打开内部的线程池功能,用于执行放入定时器中的任务,同时新开一个线程,循环等待任务到来后送入线程池中执行。
bool Run() {bool ret = thread_pool_.Start();if (!ret) {return false;}std::thread([this]() { RunLocal(); }).detach();return true;}void RunLocal() {while (running_) {std::unique_lock<std::mutex> lock(mutex_);if (queue_.empty()) {cond_.wait(lock);continue;}auto s = queue_.top();auto diff = s.time_point_ - std::chrono::high_resolution_clock::now();if (std::chrono::duration_cast<std::chrono::milliseconds>(diff).count() > 0) {cond_.wait_for(lock, diff);continue;} else {queue_.pop();lock.unlock();thread_pool_.Run(std::move(s.func_));}}}
关闭定时器功能
这里是使用running_标志位控制,标志位为false,调度线程的循环就会自动退出,就不会继续等待任务执行。
void Stop() {running_ = false;cond_.notify_all();}
在某一时间点执行任务
根据时间戳构造InternalS,放入队列中:
template <typename F, typename... Args>void AddFuncAtTimePoint(const std::chrono::time_point<std::chrono::high_resolution_clock>& time_point, F&& f,Args&&... args) {InternalS s;s.time_point_ = time_point;s.func_ = std::bind(std::forward<F>(f), std::forward<Args>(args)...);std::unique_lock<std::mutex> lock(mutex_);queue_.push(s);cond_.notify_all();}
在某段时间后执行任务
根据当前时间加上时间段构造出时间戳从而构造InternalS,放入队列中:
template <typename R, typename P, typename F, typename... Args>void AddFuncAfterDuration(const std::chrono::duration<R, P>& time, F&& f, Args&&... args) {InternalS s;s.time_point_ = std::chrono::high_resolution_clock::now() + time;s.func_ = std::bind(std::forward<F>(f), std::forward<Args>(args)...);std::unique_lock<std::mutex> lock(mutex_);queue_.push(s);cond_.notify_all();}
循环执行任务
首先为这个循环任务生成标识ID,外部可以通过ID来取消此任务继续执行,代码如下,内部以类似递归的方式循环执行任务。
template <typename R, typename P, typename F, typename... Args>int AddRepeatedFunc(int repeat_num, const std::chrono::duration<R, P>& time, F&& f, Args&&... args) {int id = GetNextRepeatedFuncId();repeated_id_state_map_.Emplace(id, RepeatedIdState::kRunning);auto tem_func = std::bind(std::forward<F>(f), std::forward<Args>(args)...);AddRepeatedFuncLocal(repeat_num - 1, time, id, std::move(tem_func));return id;}int GetNextRepeatedFuncId() { return repeated_func_id_++; }template <typename R, typename P, typename F>void AddRepeatedFuncLocal(int repeat_num, const std::chrono::duration<R, P>& time, int id, F&& f) {if (!this->repeated_id_state_map_.IsKeyExist(id)) {return;}InternalS s;s.time_point_ = std::chrono::high_resolution_clock::now() + time;auto tem_func = std::move(f);s.repeated_id = id;s.func_ = [this, &tem_func, repeat_num, time, id]() {tem_func();if (!this->repeated_id_state_map_.IsKeyExist(id) || repeat_num == 0) {return;}AddRepeatedFuncLocal(repeat_num - 1, time, id, std::move(tem_func));};std::unique_lock<std::mutex> lock(mutex_);queue_.push(s);lock.unlock();cond_.notify_all();}
取消循环任务的执行
定时器内部有repeated_id_state_map 数据结构,用于存储循环任务的ID,当取消任务执行时,将此ID从repeatedid_state_map中移除,循环任务就会自动取消。
void CancelRepeatedFuncId(int func_id) { repeated_id_state_map_.EraseKey(func_id); }
简单的测试代码
void TestTimerQueue() {TimerQueue q;q.Run();for (int i = 5; i < 15; ++i) {q.AddFuncAfterDuration(std::chrono::seconds(i + 1), [i]() { std::cout << "this is " << i << std::endl; });q.AddFuncAtTimePoint(std::chrono::high_resolution_clock::now() + std::chrono::seconds(1),[i]() { std::cout << "this is " << i << " at " << std::endl; });}int id = q.AddRepeatedFunc(10, std::chrono::seconds(1), []() { std::cout << "func " << std::endl; });std::this_thread::sleep_for(std::chrono::seconds(4));q.CancelRepeatedFuncId(id);std::this_thread::sleep_for(std::chrono::seconds(30));q.Stop();}
