? ? ? ? ?muduo的TimerQueue是基于timerfd_create實現，這樣超時很容易和epoll結合起來。等待超時事件保存在set集合中，注意set集合的有序性，從小到大排列，整個對TimerQueue的處理也就是對set集合的操作。實現TimerQueue用了3個set，分別是等待超時事件set，活躍事件set，被撤銷定時set。主要是STL的一些操作。

TimerQueue.h

// Copyright 2010, Shuo Chen. All rights reserved. // http://code.google.com/p/muduo/ // // Use of this source code is governed by a BSD-style license // that can be found in the License file.// Author: Shuo Chen (chenshuo at chenshuo dot com) // // This is an internal header file, you should not include this.#ifndef MUDUO_NET_TIMERQUEUE_H #define MUDUO_NET_TIMERQUEUE_H#include <set> #include <vector>#include "muduo/base/Mutex.h" #include "muduo/base/Timestamp.h" #include "muduo/net/Callbacks.h" #include "muduo/net/Channel.h"namespace muduo { namespace net {class EventLoop; class Timer; class TimerId;/// /// A best efforts timer queue. /// No guarantee that the callback will be on time. /// class TimerQueue : noncopyable {public:explicit TimerQueue(EventLoop* loop);~TimerQueue();////// Schedules the callback to be run at given time,/// repeats if @c interval > 0.0.////// Must be thread safe. Usually be called from other threads./* * 用于注冊定時任務* @param cb, 超時調用的回調函數* @param when，超時時間(絕對時間)* @interval，是否是周期性超時任務*/TimerId addTimer(TimerCallback cb,Timestamp when,double interval);/* 取消定時任務，每個定時任務都有對應的TimerId，這是addTimer返回給調用者的 */void cancel(TimerId timerId);private:// FIXME: use unique_ptr<Timer> instead of raw pointers.// This requires heterogeneous comparison lookup (N3465) from C++14// so that we can find an T* in a set<unique_ptr<T>>./* * 主要用于刪除操作，通過TimerId找到Timer*，再通過Timer*找到在timers_中的位置，將期刪除* 覺得可以省略*/typedef std::pair<Timestamp, Timer*> Entry;typedef std::set<Entry> TimerList;typedef std::pair<Timer*, int64_t> ActiveTimer;typedef std::set<ActiveTimer> ActiveTimerSet;void addTimerInLoop(Timer* timer);void cancelInLoop(TimerId timerId);// called when timerfd alarms/* 當timerfd被激活時調用的回調函數，表示超時 */void handleRead();// move out all expired timers/* 從timers_中拿出所有超時的Timer* */std::vector<Entry> getExpired(Timestamp now);/* 將超時任務中周期性的任務重新添加到timers_中 */void reset(const std::vector<Entry>& expired, Timestamp now);/* 插入到timers_中 */bool insert(Timer* timer);EventLoop* loop_; /* 所屬的事件驅動循環(huán) */const int timerfd_; /* 由timerfd_create創(chuàng)建的文件描述符 */Channel timerfdChannel_; /* 用于監(jiān)聽timerfd的Channel */// Timer list sorted by expirationTimerList timers_; /* 保存所有的定時任務 */// for cancel()ActiveTimerSet activeTimers_;bool callingExpiredTimers_; /* atomic */ActiveTimerSet cancelingTimers_; //保存被取消的定時器 };} // namespace net } // namespace muduo #endif // MUDUO_NET_TIMERQUEUE_H

TimerQueue.cc

// Copyright 2010, Shuo Chen. All rights reserved. // http://code.google.com/p/muduo/ // // Use of this source code is governed by a BSD-style license // that can be found in the License file.// Author: Shuo Chen (chenshuo at chenshuo dot com)#ifndef __STDC_LIMIT_MACROS #define __STDC_LIMIT_MACROS #endif#include "muduo/net/TimerQueue.h"#include "muduo/base/Logging.h" #include "muduo/net/EventLoop.h" #include "muduo/net/Timer.h" #include "muduo/net/TimerId.h"#include <sys/timerfd.h> #include <unistd.h>namespace muduo { namespace net { namespace detail {int createTimerfd() {int timerfd = ::timerfd_create(CLOCK_MONOTONIC,TFD_NONBLOCK | TFD_CLOEXEC);if (timerfd < 0){LOG_SYSFATAL << "Failed in timerfd_create";}return timerfd; }struct timespec howMuchTimeFromNow(Timestamp when)//現在距離超時時間when還有多久 {int64_t microseconds = when.microSecondsSinceEpoch()- Timestamp::now().microSecondsSinceEpoch();if (microseconds < 100){microseconds = 100;}struct timespec ts;ts.tv_sec = static_cast<time_t>(microseconds / Timestamp::kMicroSecondsPerSecond);ts.tv_nsec = static_cast<long>((microseconds % Timestamp::kMicroSecondsPerSecond) * 1000);return ts; }void readTimerfd(int timerfd, Timestamp now)//處理超時事件。超時后，timerfd變?yōu)榭勺x {uint64_t howmany;ssize_t n = ::read(timerfd, &howmany, sizeof howmany);LOG_TRACE << "TimerQueue::handleRead() " << howmany << " at " << now.toString();if (n != sizeof howmany){LOG_ERROR << "TimerQueue::handleRead() reads " << n << " bytes instead of 8";} }//timerfd是時間對于的文件描述符 void resetTimerfd(int timerfd, Timestamp expiration)//重新設置定時器 {// wake up loop by timerfd_settime()struct itimerspec newValue;struct itimerspec oldValue;memZero(&newValue, sizeof newValue);memZero(&oldValue, sizeof oldValue);newValue.it_value = howMuchTimeFromNow(expiration);int ret = ::timerfd_settime(timerfd, 0, &newValue, &oldValue);if (ret){LOG_SYSERR << "timerfd_settime()";} }} // namespace detail } // namespace net } // namespace muduousing namespace muduo; using namespace muduo::net; using namespace muduo::net::detail;TimerQueue::TimerQueue(EventLoop* loop): loop_(loop),timerfd_(createTimerfd()),timerfdChannel_(loop, timerfd_),timers_(),callingExpiredTimers_(false) {timerfdChannel_.setReadCallback(std::bind(&TimerQueue::handleRead, this));// we are always reading the timerfd, we disarm it with timerfd_settime.timerfdChannel_.enableReading(); }TimerQueue::~TimerQueue() {timerfdChannel_.disableAll();timerfdChannel_.remove();::close(timerfd_);// do not remove channel, since we're in EventLoop::dtor();for (const Entry& timer : timers_){delete timer.second;} }TimerId TimerQueue::addTimer(TimerCallback cb,Timestamp when,double interval) {Timer* timer = new Timer(std::move(cb), when, interval);/* * 在自己所屬線程調用addTimerInLoop函數 */loop_->runInLoop(std::bind(&TimerQueue::addTimerInLoop, this, timer));return TimerId(timer, timer->sequence()); }void TimerQueue::cancel(TimerId timerId) {loop_->runInLoop(std::bind(&TimerQueue::cancelInLoop, this, timerId)); }/* 向計時器隊列中添加超時事件 */ void TimerQueue::addTimerInLoop(Timer* timer) {loop_->assertInLoopThread();bool earliestChanged = insert(timer);//返回true，說明timer被添加到set的頂部，作為新的根節(jié)點，需要更新timerfd的激活時間// 只有在計時器為空的時候或者新加入的計時器的最早觸發(fā)時間小于當前計時器的堆頂的最小值// 才需要用最近時間去更新if (earliestChanged){resetTimerfd(timerfd_, timer->expiration());} }void TimerQueue::cancelInLoop(TimerId timerId) {loop_->assertInLoopThread();assert(timers_.size() == activeTimers_.size());ActiveTimer timer(timerId.timer_, timerId.sequence_);ActiveTimerSet::iterator it = activeTimers_.find(timer);if (it != activeTimers_.end())//要取消的在當前激活的Timer集合中{size_t n = timers_.erase(Entry(it->first->expiration(), it->first));//在timers_中取消assert(n == 1); (void)n;delete it->first; // FIXME: no delete pleaseactiveTimers_.erase(it);//在activeTimers_中取消}else if (callingExpiredTimers_)//如果正在執(zhí)行超時定時器的回調函數，則加入到cancelingTimers集合中{cancelingTimers_.insert(timer);}assert(timers_.size() == activeTimers_.size()); }/* * 當定時器超時，保存timerfd的Channel激活，調用回調函數*/ void TimerQueue::handleRead() {loop_->assertInLoopThread();Timestamp now(Timestamp::now());readTimerfd(timerfd_, now);/* 從定時任務set中拿出所有超時任務 */std::vector<Entry> expired = getExpired(now);callingExpiredTimers_ = true;cancelingTimers_.clear();// safe to callback outside critical section/* 調用超時的事件回調函數 */for (const Entry& it : expired){it.second->run();}callingExpiredTimers_ = false;reset(expired, now); }/** 重新整理時間set中的任務，將所有超時的任務都拿出，然后調用其回調函數*/ std::vector<TimerQueue::Entry> TimerQueue::getExpired(Timestamp now) {assert(timers_.size() == activeTimers_.size());std::vector<Entry> expired;Entry sentry(now, reinterpret_cast<Timer*>(UINTPTR_MAX));//返回第一個大于等于now的迭代器，小于now的都已經超時//lower_bound( begin,end,num)：從數組的begin位置到end-1位置二分查找第一個大于或等于num的數字，找到返回該數字的地址//lower_bound(val):返回容器中第一個值【大于或等于】val的元素的iterator位置TimerList::iterator end = timers_.lower_bound(sentry);assert(end == timers_.end() || now < end->first);/* back_inserter：容器適配器，將數據插入到參數的尾部 *///一個序列（sequence）拷貝到一個容器（container）中去，通常用std::copy算法std::copy(timers_.begin(), end, back_inserter(expired));timers_.erase(timers_.begin(), end); //從timers_中移除for (const Entry& it : expired){ActiveTimer timer(it.second, it.second->sequence());size_t n = activeTimers_.erase(timer); //從activeTimers_中移除assert(n == 1); (void)n;}assert(timers_.size() == activeTimers_.size());return expired; }/* * 調用完所有超時的回調函數后，需要對這些超時任務進行整理* 將周期性的定時任務重新添加到set中*/ void TimerQueue::reset(const std::vector<Entry>& expired, Timestamp now) {Timestamp nextExpire;for (const Entry& it : expired){ActiveTimer timer(it.second, it.second->sequence());if (it.second->repeat() /* 是否是周期性的定時任務 */&& cancelingTimers_.find(timer) == cancelingTimers_.end()){/* 重新計算超時時間 */it.second->restart(now);/* 重新添加到set中 */insert(it.second);}else{// FIXME move to a free listdelete it.second; // FIXME: no delete please}}/* 計算下次timerfd被激活的時間 */if (!timers_.empty()){nextExpire = timers_.begin()->second->expiration();//set從小到大排序}/* 設置 */if (nextExpire.valid())//時間是有效的{resetTimerfd(timerfd_, nextExpire);} }bool TimerQueue::insert(Timer* timer) {loop_->assertInLoopThread();assert(timers_.size() == activeTimers_.size());bool earliestChanged = false;/* 獲取timer的UTC時間戳，和timer組成std::pair<Timestamp, Timer*> */Timestamp when = timer->expiration();/* timers_begin()是set頂層元素（紅黑樹根節(jié)點），是超時時間最近的Timer* */TimerList::iterator it = timers_.begin();/* 如果要添加的timer的超時時間比timers_中的超時時間近，更改新的超時時間 */if (it == timers_.end() || when < it->first){earliestChanged = true;}{/* 添加到定時任務的set中 */std::pair<TimerList::iterator, bool> result = timers_.insert(Entry(when, timer));assert(result.second); (void)result;}{/* 同時也添加到activeTimers_中，用于刪除時查找操作 */std::pair<ActiveTimerSet::iterator, bool> result = activeTimers_.insert(ActiveTimer(timer, timer->sequence()));assert(result.second); (void)result;}assert(timers_.size() == activeTimers_.size());return earliestChanged; }

?

總結

以上是生活随笔為你收集整理的muduo之TimerQueue的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。