上次說到，基于排序鏈表的定時器存在一個問題：添加定時器的效率偏低。這次我們用時間輪來解決該問題。
如圖就是一個時間輪：
在時間輪內，指針指向輪子上的一個槽。它以恒定的速率順時針轉動。沒轉動一步就指向下一個槽，每次轉動稱之為一個tick。一個滴答的時間稱為時間輪的槽間隔si（slot interval），它實際上就是心搏時間。時間輪共有N個槽，因此它運轉一周的時間是N*si。每個槽指向一個定時器鏈表，每條鏈表上的定時器具有相同的特征：它們的定時時間相差N*si的整數倍。時間輪正式利用這個關系將定時器散列到不同的鏈表中。加入現在指針指向槽cs，我們要添加一個定時時間為ti的定時器，則該定時器將被插入槽ts（timer slot）對應的鏈表中：
ts = (cs + (ti / si)) % N
基于排序鏈表的定時器使用唯一的鏈表來管理所有定時器，所以插入操作的效率隨著定時器數目的增多而降低。而時間輪使用哈希表的思想，將定時器散列到不同的鏈表上。這樣每條鏈表上的定時器數目都將明顯少于原來的排序鏈表上的定時器數目，插入操作的效率基本不受定時器數目的影響。
很顯然，對時間輪而言，要提高定時精度，就要使si值足夠??；要提高執行效率，則要求N值足夠大。
上圖描述的是一個簡單的時間輪，僅僅一個輪子。而復雜的時間輪可能有多個輪子，不同輪子擁有不同的粒度。
下面是一個簡單時間輪的實現代碼：
#ifndef TIME_WHEEL_TIMER_H#define TIME_WHEEL_TIMER_H#include <time.h>#include <netinet/in.h>#include <stdio.h>#include <assert.h>const int BUFFER_SIZE = 1024;class tw_timer;//綁定socket和定時器struct client_data { sockaddr_in addr_; int sockfd_; char buf_[BUFFER_SIZE]; tw_timer* timer_;};//定時器類class tw_timer {public: tw_timer(int rot, int ts) : next_(NULL), PRev_(NULL), rotation_(rot), time_slot_(ts) {} public: void (*timeout_callback_)(client_data*); //定時器回調函數public: int rotation_; //記錄定時器在時間輪轉多少圈后生效，因為有的定時值比較大 int time_slot_; //記錄定時器對應于時間輪上的哪個槽（對應的鏈表） client_data *user_data_; //客戶數據 tw_timer* next_; //指向上一個定時器 tw_timer* prev_; //指向下一個定時器};class time_wheel {public: time_wheel() : cur_slot_(0) { memset(slots_, 0, sizeof(slots_)); //清零每個槽指針 } ~time_wheel(){ //遍歷每個槽，并銷毀其中的定時器 for(int i=0; i<DEFAULT_SLOTS_NUM; ++i){ tw_timer* tmp = slots_[i]; while(tmp != NULL){ slots_[i] = tmp->next_; delete tmp; tmp = slots_[i]; } } }public: tw_timer* add_timer(int timeout); tw_timer* adjust_timer(tw_timer* timer, int timeout); void del_timer(tw_timer* timer); void tick();private: static const int DEFAULT_SLOTS_NUM = 60; static const int SI = 1; tw_timer* slots_[DEFAULT_SLOTS_NUM]; int cur_slot_;};//根據定時值timeout創建一個定時器，并把它插入合適的槽中tw_timer* time_wheel::add_timer(int timeout){ if(timeout < 0) return NULL; //下面根據待插入定時器的超時值計算它將在時間輪轉動多少個滴答后被觸發，并將該滴答數存儲于變量ticks中。 //如果待插入定時器的超時值小于時間輪的槽間隔SI，則將ticks折合為1，下一次它就被觸發。否則將ticks向下折合為timeout/SI int ticks = 0; if(timeout < SI) ticks = 1; else ticks = timeout / SI; //計算待插入定時器在時間輪轉多少圈后被觸發 int rotation = ticks / DEFAULT_SLOTS_NUM; //計算待插入的定時器應該被插入哪個槽中 int ts = (cur_slot_ + (ticks % DEFAULT_SLOTS_NUM)) % DEFAULT_SLOTS_NUM; //創建新的定時器，它在時間輪轉動rotation圈之后被觸發，且位于第ts個槽上 tw_timer* timer = new tw_timer(rotation, ts); //如果第ts個槽中無任何定時器，則把新建的定時器插入其中，并將該定時器設置為該槽的頭結點 if(slots_[ts] == NULL){ printf("add timer, rotation is %d, ts is %d, cur_slot_ is %d/n", rotation, ts, cur_slot_); slots_[ts] = timer; } else{ //否則，將定時器插入第ts個槽中 timer->next_ = slots_[ts]; slots_[ts]->prev_ = timer; slots_[ts] = timer; } return timer;}//調整定時器，延長壽命tw_timer* time_wheel::adjust_timer(tw_timer* timer, int timeout){ assert(timer != NULL && timeout >= 0); printf("adjust timer/n"); del_timer(timer); //延長壽命我們需要刪掉之前的，從新添加一個新的 return add_timer(timeout);}//刪除目標定時器timervoid time_wheel::del_timer(tw_timer* timer){ if(timer == NULL) return ; int ts = timer->time_slot_; //slots_[ts]是目標定時器所在槽的頭結點。如果目標定時器就是該頭結點，則需要重置第ts個槽的頭結點 if(timer == slots_[ts]){ slots_[ts] = slots_[ts]->next_; if(slots_[ts] != NULL) slots_[ts]->prev_ = NULL; delete timer; } else{ timer->prev_->next_ = timer->next_; if(timer->next_ != NULL){ timer->next_->prev_ = timer->prev_; } delete timer; }}//SI時間到后，調用該函數，先檢驗時間輪對應的槽的所有timer是否到期了，進行相應的處理。然后時間輪向前滾動一個槽的間隔。void time_wheel::tick(){ tw_timer* tmp = slots_[cur_slot_]; //取得時間輪上當前槽的頭結點 printf("current slot is %d/n", cur_slot_); while(tmp != NULL){ printf("tick the timer once/n"); //如果定時器的rotation值大于0，則它在這一輪補齊作用，它壽命還長著呢 if(tmp->rotation_ > 0){ tmp->rotation_--; tmp = tmp->next_; //繼續查找下一個timer } else{ //否則，說明定時器已經到期，于是執行定時任務，然后刪除該定時器 tmp->timeout_callback_(tmp->user_data_); if(tmp == slots_[cur_slot_]){ printf("delete header in cur_slot/n"); slots_[cur_slot_] = tmp->next_; delete tmp; if(slots_[cur_slot_] != NULL) slots_[cur_slot_]->prev_ == NULL; tmp = slots_[cur_slot_]; } else{ tmp->prev_->next_ = tmp->next_; if(tmp->next_ != NULL) tmp->next_->prev_ = tmp->prev_; tw_timer* tmp2 = tmp->next_; delete tmp; tmp = tmp2; } } } //更新時間輪的當前槽，向前走一步，以反映時間輪的轉動 cur_slot_ = ++cur_slot_ % DEFAULT_SLOTS_NUM; //similar to cycle queue}#endif下面是測試代碼，類似上篇博客中升序鏈表的測試代碼，僅有部分不同：
#include <sys/types.h>#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#include <assert.h>#include <stdio.h>#include <signal.h>#include <unistd.h>#include <errno.h>#include <string.h>#include <fcntl.h>#include <stdlib.h>#include <sys/epoll.h>#include <pthread.h>#include "time_wheel_timer.h"#define FD_LIMIT 65535#define MAX_EVENT_NUMBER 1024#define TIME_SLOT 5static int pipefd[2];static time_wheel timer_lst;static int epollfd = 0;int setnonblocking( int fd ){ int old_option = fcntl( fd, F_GETFL ); int new_option = old_option | O_NONBLOCK; fcntl( fd, F_SETFL, new_option ); return old_option;}void addfd(int fd ){ epoll_event event; event.data.fd = fd; event.events = EPOLLIN | EPOLLET; epoll_ctl( epollfd, EPOLL_CTL_ADD, fd, &event ); setnonblocking( fd );}void sig_handler( int sig ){ int save_errno = errno; int msg = sig; send( pipefd[1], ( char* )&msg, 1, 0 ); errno = save_errno;}void addsig( int sig ){ struct sigaction sa; memset( &sa, '/0', sizeof( sa ) ); sa.sa_handler = sig_handler; sa.sa_flags |= SA_RESTART; sigfillset( &sa.sa_mask ); assert( sigaction( sig, &sa, NULL ) != -1 );}void timer_handler(){ timer_lst.tick(); alarm( TIME_SLOT );}void cb_func( client_data* user_data ){ epoll_ctl( epollfd, EPOLL_CTL_DEL, user_data->sockfd_, 0 ); assert( user_data ); close( user_data->sockfd_ ); printf( "close fd %d/n", user_data->sockfd_ );}int main( int argc, char* argv[] ){ if( argc <= 2 ) { printf( "usage: %s ip_address port_number/n", basename( argv[0] ) ); return 1; } const char* ip = argv[1]; int port = atoi( argv[2] ); int ret = 0; struct sockaddr_in address; bzero( &address, sizeof( address ) ); address.sin_family = AF_INET; inet_pton( AF_INET, ip, &address.sin_addr ); address.sin_port = htons( port ); int listenfd = socket( PF_INET, SOCK_STREAM, 0 ); assert( listenfd >= 0 ); int on = 1; ret = setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)); assert(ret != -1); ret = bind( listenfd, ( struct sockaddr* )&address, sizeof( address ) ); assert( ret != -1 ); ret = listen( listenfd, 5 ); assert( ret != -1 ); epoll_event events[ MAX_EVENT_NUMBER ]; epollfd = epoll_create( 5 ); assert( epollfd != -1 ); addfd(listenfd ); ret = socketpair( PF_UNIX, SOCK_STREAM, 0, pipefd ); assert( ret != -1 ); setnonblocking( pipefd[1] ); addfd(pipefd[0] ); // add all the interesting signals here addsig( SIGALRM ); addsig( SIGTERM ); bool stop_server = false; client_data* users = new client_data[FD_LIMIT]; bool timeout = false; alarm( TIME_SLOT ); while( !stop_server ) { int number = epoll_wait( epollfd, events, MAX_EVENT_NUMBER, -1 ); if ( ( number < 0 ) && ( errno != EINTR ) ) { printf( "epoll failure/n" ); break; } for ( int i = 0; i < number; i++ ) { int sockfd = events[i].data.fd; if( sockfd == listenfd ) { struct sockaddr_in client_address; socklen_t client_addrlength = sizeof( client_address ); int connfd = accept( listenfd, ( struct sockaddr* )&client_address, &client_addrlength ); addfd(connfd); users[connfd].addr_ = client_address; users[connfd].sockfd_ = connfd; tw_timer* timer = timer_lst.add_timer(3 * TIME_SLOT); timer->user_data_ = &users[connfd]; timer->timeout_callback_ = cb_func; users[connfd].timer_ = timer; } else if( ( sockfd == pipefd[0] ) && ( events[i].events & EPOLLIN ) ) { int sig; char signals[1024]; ret = recv( pipefd[0], signals, sizeof( signals ), 0 ); if( ret == -1 ) { // handle the error continue; } else if( ret == 0 ) { continue; } else { for( int i = 0; i < ret; ++i ) { switch( signals[i] ) { case SIGALRM: { timeout = true; break; } case SIGTERM: { stop_server = true; } } } } } else if( events[i].events & EPOLLIN ) { memset( users[sockfd].buf_, '/0', BUFFER_SIZE ); ret = recv( sockfd, users[sockfd].buf_, BUFFER_SIZE-1, 0 ); printf( "get %d bytes of client data %s from %d/n", ret, users[sockfd].buf_, sockfd ); tw_timer* timer = users[sockfd].timer_; if( ret < 0 ) { if( errno != EAGAIN ) { cb_func( &users[sockfd] ); if( timer ) { timer_lst.del_timer( timer ); } } } else if( ret == 0 ) { cb_func( &users[sockfd] ); if( timer ) { timer_lst.del_timer( timer ); } } else { //send( sockfd, users[sockfd].buf, BUFFER_SIZE-1, 0 ); if( timer ) { //下面這些注釋代碼是和上篇博客升序鏈表不同的地方之一 // time_t cur = time( NULL ); //timer->expire = cur + 3 * TIMESLOT; // printf( "adjust timer once/n" ); //timer_lst.adjust_timer( timer ); tw_timer* new_timer = timer_lst.adjust_timer(timer, 3*TIME_SLOT); new_timer->user_data_ = &users[sockfd]; new_timer->timeout_callback_ = cb_func; users[sockfd].timer_ = new_timer; } } } else { // others } } if( timeout ) { timer_handler(); timeout = false; } } close( listenfd ); close( pipefd[1] ); close( pipefd[0] ); close( epollfd ); delete [] users; return 0;}對于時間輪而言，添加一個定時器的時間復雜度是O(1)，刪除一個定時器的時間復雜度也是O(1)（因為是雙向鏈表直接利用prev指針），執行一個定時器的時間復雜度是O(n)（遍歷某個槽的鏈表所有節點，因為有的節點輪數不是當前輪，所以我們不能憑借類似升序鏈表那樣只遍歷部分鏈表就知道后面的節點時間未到）。但實際上執行一個定時器任務的效率比O(n)好的多，因為時間輪將所有定時器散列到不同的鏈表上。時間輪的槽越多，每條鏈表上定時器數量越少。當采用多個輪子實現時間輪，執行一個定時器的時間復雜度接近O(1)。
新聞熱點
疑難解答
