c/c++用一个源文件，一个头文件实现的 优先队列定时器 源码，包含测试代码，可运行

#include <queue> #include <deque> #include <set> #include <stdint.h> #include <unistd.h> #include <pthread.h> #include <memory> #include <sys/time.h> #include <cstring> #include "qtimer.h" using namespace std; typedef struct _qtimer_t_ { TimerCtx* ctx; time_t interval; struct timespec expired; uint8_t* pData; int pData_Len; Callback_OnExpire pfunc; uint32_t timer_id; }Qtimer; class Qcompare { public: bool operator()(const shared_ptr<Qtimer> &t1, const shared_ptr<Qtimer> &t2) { return (t1->expired.tv_sec > t2->expired.tv_sec) || (t1->expired.tv_sec == t2->expired.tv_sec && t1->expired.tv_nsec > t2->expired.tv_nsec); //小根堆 } }; class TimerCtx { public: TimerCtx() { pthread_mutex_init(&mutex_timer, NULL); pthread_cond_init(&cond_timer, NULL); run_status = 1; id_counter = 1; } ~TimerCtx() { pthread_mutex_destroy(&mutex_timer); pthread_cond_destroy(&cond_timer); } public: pthread_t pthd; priority_queue<shared_ptr<Qtimer>, deque<shared_ptr<Qtimer>>, Qcompare> timer_queue; set<uint32_t> id_set; //已分配的timer id 集, 目的只要是为了方便去重, 因为优先队列的变量效率低。 volatile uint8_t run_status; //0:定时器线程停止, 服务退出; 1:定时器线程运行 pthread_mutex_t mutex_timer; pthread_cond_t cond_timer; //private: shared_ptr<Qtimer> new_init_timer(time_t interval, Callback_OnExpire pfunc, uint8_t* pData, int len); uint32_t allocate_timer_id(); uint32_t check_timer_id(uint32_t timer_id); uint32_t id_counter; //用于分配timer_id set<uint32_t> remove_set; //需要移除的timer id 集 }; uint32_t TimerCtx::allocate_timer_id() { uint32_t id; while(1) { id = id_counter++; id = (id==0)?id_counter++:id; //id!=0 if(id_set.find(id) != id_set.end()) continue; else break; } return id; } uint32_t TimerCtx::check_timer_id(uint32_t timer_id) { if(id_set.find(timer_id) != id_set.end()) { return -1; } return timer_id; } void* timer_thread(void* p) { TimerCtx* tm = (TimerCtx *)p; while(tm->run_status) { pthread_mutex_lock(&tm->mutex_timer); if(tm->timer_queue.empty()) { //TODO: wait signel struct timespec outtime; outtime.tv_sec = time(NULL) + 1; outtime.tv_nsec = 0; pthread_cond_timedwait(&(tm->cond_timer), &(tm->mutex_timer), &outtime); pthread_mutex_unlock(&tm->mutex_timer); continue; } struct timeval now; gettimeofday(&now, NULL); shared_ptr<Qtimer> timer = tm->timer_queue.top(); if(timer->expired.tv_sec < now.tv_sec || (timer->expired.tv_sec == now.tv_sec && timer->expired.tv_nsec <= (now.tv_usec*1000))) { tm->timer_queue.pop(); tm->id_set.erase(timer->timer_id); if(tm->remove_set.find(timer->timer_id) == tm->remove_set.end()) { pthread_mutex_unlock(&tm->mutex_timer); timer->pfunc(tm, timer->timer_id, timer->pData, timer->pData_Len); }else { tm->remove_set.erase(timer->timer_id); //timer->pData需要用户自己释放 pthread_mutex_unlock(&tm->mutex_timer); } //用户数据的拷贝, 由timermanager服务释放 if(timer->pData) { delete[] timer->pData; timer->pData = nullptr; } timer = nullptr; continue; } //TODO: wait (*) seconds pthread_cond_timedwait(&(tm->cond_timer), &(tm->mutex_timer), &timer->expired); pthread_mutex_unlock(&tm->mutex_timer); } return NULL; } TimerCtx* q_timer_ctx_init() { TimerCtx* q_tm = nullptr; q_tm = new TimerCtx(); if(!q_tm) { return nullptr; } int res = pthread_create(&q_tm->pthd, NULL, timer_thread, q_tm); if(res < 0) { printf("q_timer_init pthread_create() failed, res:%d\n", res); delete q_tm; return nullptr; } sleep(1); return q_tm; } shared_ptr<Qtimer> TimerCtx::new_init_timer(time_t interval, Callback_OnExpire pfunc, uint8_t* pData, int len) { shared_ptr<Qtimer> timer = shared_ptr<Qtimer>(new Qtimer()); if(timer == nullptr) { printf(" new Qtimer failed\n"); return nullptr; } timer->ctx = this; timer->interval = interval; timer->pfunc = pfunc; timer->pData = new uint8_t[len + 1](); if(timer->pData == nullptr) { printf(" new pData failed\n"); timer = nullptr; return nullptr; } //用户数据的拷贝, 需由timermanager服务释放 memcpy(timer->pData, pData, len); timer->pData_Len = len; struct timeval now; gettimeofday(&now, NULL); timer->expired.tv_sec = now.tv_sec+interval; timer->expired.tv_nsec = now.tv_usec * 1000; return timer; } int q_add_timer(TimerCtx* ctx, time_t interval, Callback_OnExpire pfunc, uint8_t* pData, int len) { if(ctx->run_status == 0) return -1; shared_ptr<Qtimer> timer = ctx->new_init_timer(interval, pfunc, pData, len); if(timer == nullptr) { printf(" new Qtimer failed\n"); return -1; } pthread_mutex_lock(&ctx->mutex_timer); timer->timer_id = ctx->allocate_timer_id(); ctx->timer_queue.push(timer); ctx->id_set.insert(timer->timer_id); pthread_cond_signal(&ctx->cond_timer); pthread_mutex_unlock(&ctx->mutex_timer); return timer->timer_id; } int q_add_timer(TimerCtx* ctx, uint32_t timer_id, time_t interval, Callback_OnExpire pfunc, uint8_t* pData, int len) { if(ctx->run_status == 0) return -1; shared_ptr<Qtimer> timer = ctx->new_init_timer(interval, pfunc, pData, len); if(timer == nullptr) { printf(" new Qtimer failed\n"); return -1; } pthread_mutex_lock(&ctx->mutex_timer); timer->timer_id = ctx->check_timer_id(timer_id); if(timer->timer_id < 0) { pthread_mutex_unlock(&ctx->mutex_timer); delete[] timer->pData; timer = nullptr; return -1; } ctx->timer_queue.push(timer); ctx->id_set.insert(timer->timer_id); pthread_cond_signal(&ctx->cond_timer); pthread_mutex_unlock(&ctx->mutex_timer); return timer->timer_id; } int q_remove_timer(TimerCtx* ctx, uint32_t id) { pthread_mutex_lock(&ctx->mutex_timer); if(ctx->id_set.find(id) == ctx->id_set.end()) { pthread_mutex_unlock(&ctx->mutex_timer); return 0; } ctx->remove_set.insert(id); pthread_mutex_unlock(&ctx->mutex_timer); return 0; } size_t q_get_timernum(TimerCtx* ctx) { return ctx->timer_queue.size(); } int q_timer_release(TimerCtx* ctx) { ctx->run_status = 0; pthread_mutex_lock(&ctx->mutex_timer); pthread_cond_signal(&ctx->cond_timer); pthread_mutex_unlock(&ctx->mutex_timer); pthread_join(ctx->pthd, NULL); if(ctx->timer_queue.size()>0) { do { shared_ptr<Qtimer> timer = ctx->timer_queue.top(); if(timer && timer->pData) { delete[] timer->pData; timer->pData = nullptr; } ctx->timer_queue.pop(); }while(!ctx->timer_queue.empty()); } if(ctx) delete ctx; return 0; }