优先队列实现定时器

  1 #include <iostream>
  2 #include <sys/time.h>
  3 #include <unistd.h>
  4 #include <thread>
  5 #include <vector>
  6 #include <map>
  7 #include <memory>
  8 #include <mutex>
  9 #include <stdlib.h>
 10 #include <algorithm>
 11 #include <queue>
 12 
 13 using namespace std;
 14 
 15 #define THREAD_COUNT 4
 16 
 17 typedef struct TimerInfo
 18 {
 19     int id;
 20     long expired;     //超时时间戳(ms)
 21     bool is_working;  //用于判断该定时器是否有效
 22     bool is_repeat;  // 是否为循环定时器标识
 23     long interval; //循环定时器循环间隔
 24 
 25 } TimerInfo;
 26 
 27 // 排序函数
 28 class comp
 29 {
 30 public:
 31     bool operator()(const shared_ptr<TimerInfo> &t1, const shared_ptr<TimerInfo> &t2)
 32     {
 33         if(t1.get()->expired > t2.get()->expired)
 34             return true;
 35         else
 36             return false;
 37     }
 38 };
 39 
 40 
 41 typedef struct TimerMng
 42 {
 43     int id;
 44     
 45     priority_queue<shared_ptr<TimerInfo>, vector<shared_ptr<TimerInfo> >, comp> timer_queue;   // 优先队列
 46     map<int, shared_ptr<TimerInfo> > timer_map;   //id到定时器的映射 用于根据id取消定时器
 47 
 48     mutex mtx;
 49 
 50     struct timeval now;
 51 
 52     TimerMng()
 53     {
 54         id = 0;
 55     }
 56 
 57 } TimerMng;
 58 
 59 TimerMng t_mng;
 60 
 61 void add_timer(bool is_repeat, long expired, long interval)
 62 {
 63     TimerInfo *t_info = new TimerInfo();
 64     t_info->is_repeat = is_repeat;
 65     t_info->expired = expired;
 66     t_info->interval = interval;
 67     t_info->is_working = true;
 68 
 69     shared_ptr<TimerInfo> timer_info(t_info);
 70 
 71     lock_guard<mutex> guard(t_mng.mtx);
 72     int now_id = t_mng.id++;
 73     t_info->id = now_id;
 74 
 75     t_mng.timer_queue.push(timer_info);
 76 
 77     t_mng.timer_map.insert(pair<int, shared_ptr<TimerInfo> >(now_id, timer_info) );
 78 
 79 }
 80 
 81 void update_time()
 82 {
 83     gettimeofday(&t_mng.now, NULL);
 84 }
 85 
 86 void execute_timer()
 87 {
 88     long now_usec = t_mng.now.tv_sec * 1000 + t_mng.now.tv_usec / 1000;
 89 
 90     while(true) 
 91     {
 92         lock_guard<mutex> guard(t_mng.mtx);
 93 
 94         if (t_mng.timer_queue.empty())
 95             break;
 96 
 97         shared_ptr<TimerInfo> first = t_mng.timer_queue.top();
 98 
 99         if (first.get()->expired > now_usec )
100             break;
101 
102         if (first.get()->is_working)
103         {
104             // do something here
105             cout << "timer execute succ, now: " << now_usec <<  " id: " << first.get()->id << " " << "expired: " << first.get()->expired << endl;
106         }
107 
108         map<int, shared_ptr<TimerInfo> >::iterator map_iter = t_mng.timer_map.find( first.get()->id);
109 
110         // 从map中移除
111         t_mng.timer_map.erase(map_iter);
112 
113         // 从队列中移除
114         t_mng.timer_queue.pop();
115 
116     }
117 }
118 
119 
120 void timer_thread()
121 {
122     while(1)
123     {
124         update_time();
125         execute_timer();
126         // 1ms 1次循环
127         usleep(1000);
128     }
129 }
130 
131 void worker_thread()
132 {
133     srand((unsigned)time(0));
134     while(1)
135     {
136         struct timeval now;
137         gettimeofday(&now, NULL);
138 
139         int rand_sec = rand() % 5 + 1;
140         int rand_usec = rand() % 900000 + 1;
141 
142         long expired = (now.tv_sec + rand_sec) * 1000 + ( rand_usec / 1000 );
143         add_timer(false, expired, 0);
144 
145         sleep(rand() % 2 + 1);    
146     }
147 }
148 
149 
150 int main()
151 {
152     thread t1(timer_thread);
153 
154     vector<thread> v_thread;
155 
156     for (int i = 0; i < THREAD_COUNT; ++i)
157     {
158         v_thread.push_back(thread(worker_thread));
159     }
160 
161     t1.join();
162 
163     for (int i = 0; i < THREAD_COUNT; ++i)
164     {
165         v_thread[i].join();
166     }
167 }