游戲服務端定時器的實現


最近在看過一些定時器相關的資料,也讀了一些代碼,比如雲風的skynet的定時器實現,小有啟發,因此將所得整理記錄下來。  

通常,一個定時器模塊會提供以下三個接口:

  1. reg_tick(timeout, callback);
  2. unreg_tick(tick_id);
  3. update_timer();

reg_tick注冊一個tick,unreg_tick取消一個tick,update_timer更新計時器的時間,觸發其中過期的tick。前兩個接口好理解,問題是第三個接口update_timer,到底什么時候調用這個呢?看下面的說明。

一個游戲服務端需要處理客戶端的請求,也要處理定時的任務,其主循環或許如下:

1 while 1:
2     #處理網絡IO任務
3     result = select(10)
4     for fd, event in result:
5         handle(fd, event)
6     #處理定時任務
7     update_timer()

update_time這個函數的工作就是找出服務器所有過期的定時任務,並執行其對應的回調函數。最簡單的實現是,假設服務器有一個集合保存着所有注冊了的tick(一個tick就是一個定時任務,下文將不再解釋),每次更新計時器的時候,遍歷集合中的所有tick,挨個去判斷這個tick是否過期,如果是則執行其callback,可能的話還要刪掉這個已經觸發了的tick。

1 def update_timer():
2     current = get_cur_time()
3     for timer in reg_timer_list:
4         if timer.expeires > current:
5             timer.callback()

接下來分析這個實現的時間復雜度。如果使用vector來存儲tick的話,reg_tick的時間是O(1),unreg_tick的時間是O(n),update_timer的時間也是O(n)。

當然有幾種簡單的改進方法,比如改為用hash_map來存儲tick,可以將插入和刪除的時間降到O(1),但需要注意的是對於一般的hash_map如果沒有記錄前一個和后一個元素的位置,遍歷起來是會相對耗時的。因此update_timer的時間可能會需要更多,具體依賴hash_map的實現。當然也可以使用紅黑樹來存儲tick。

另一種改進方法是,依然使用占內存相對較小的vector存儲tick,不過需要維持這個vector,使其中所有tick都是按觸發時間從早到晚的依次排序。這樣一來插入和刪除的時間變為O(n),但是相對的,update_timer卻變快了,因為可以把要觸發的tick都集中到一起,攤還下來從而每一個tick的觸發時間變少了。這樣做通常是有好處的,如上文所述,相對於reg_tick和unreg_tick服務器一般會較為頻繁的調用update_timer這個函數。

其實基於維持一個有序數組的思想還可以進一步的優化,比如用觸發時間為key使用最小堆來存儲tick。這種情況插入刪除和有序數組時間復雜度一致,但是update_timer一般來說效率更高。

其實還可以進一步再優化,這次依然選擇hash_map作為存儲tick的數據結構,只不過存儲在hash_map中的key是一個時間戳,value是這個時間戳對應的所有tick列表。這樣一來每次調用update_timer這個函數不再需要遍歷整個hash_map,而是根據時間戳索引到對應的tick列表,因此update_timer的時間復雜度降至O(1)。由此而來,reg_timer、unreg_timer、update_timer三個接口的時間復雜度都降至O(1)。

1 def update_timer():
2     last = get_last_update()
3     current = get_cur_time()
4     while last <= current:
5         for timer in reg_timer_dict[last]:
6             timer.callback()
7             last += 1

到這為止了嗎?

通用性的解決方案是為了滿足大多數的情況而被使用的,在特定問題下,如果我們根據問題的獨特性進行相應的優化,通常能做的更好,這也是造輪子的意義所在。回到計時器這個問題,我們是否能優化一下hash_map空間復雜度呢?

接下來要引入的就是特定情況下的解決方案。我們可以基於一個假設進行優化,假設我們不需要注冊一個很久以后的tick。由此而引出的是分層時間輪計時器,這個也是Linux內核使用的定時器算法(skynet的定時器也是,可以讀一下skynet的代碼,只有兩百多行)。

一般使用的都是分層時間輪,朴素時間輪算法就不作說明了。之前在看資料的時候,找到了一篇文章對分層時間輪描述的很詳細,地址是見文章結尾的參考資料。

我在這里就簡單說明一下。假設這個有三個時鍾,分別是時分秒。以秒級時鍾說明,這個時鍾有60個槽,代表了一分鍾的60秒,每一個槽對應存放的是tick列表。我們有一個變量記錄下當前的秒針的位置,每當秒針走一步,則觸發對應槽的tick列表中的所有tick。假設我要注冊一個10秒鍾之后觸發的tick,則把這個tick插入到(當前秒針+10)mod 60的位置上即可。這樣60秒內的tick沒什么問題,但是大於60秒則如何處理?這時就要使用到分鍾級的時鍾了。比如說我需要注冊一個70秒后觸發的tick,通過計算可以發現這個tick是下一分鍾觸發的,因此把這個tick存放在(當前分針+1)mod 60的位置上。當秒針轉完一輪后,分針需要走一步,這個時候就把當前分針所指向的槽的所有tick都插入到秒級時鍾去。比如說剛剛的70秒后的tick,因為秒針走了一輪,這時候觸發時間變成10秒之后,因此把它插入到(當前秒針+10)mod 60的位置。時針的處理與這個類似。通過這種方法,表示一天需要的空間復雜度是60+60+24=144個槽。

Linux內核和Skynet的定時器都是使用這個方法,但是它們不是按時分秒這樣分層,而是將32bit按8/6/6/6/6/分成5個部分,也就是有5個時鍾(這里每一個時鍾被稱為Time Vector簡稱TV),原來的秒級時鍾變為2^8=256個槽。這種分層方法的空間復雜度變為256+64+64+64+64=512個槽,支持注冊最長時間的tick是256*64*64*64*64=2^32秒。

 

參考資料:

  1. 淺析 Linux 中的時間編程和實現原理,第 3 部分: Linux 內核的工作。地址:http://www.ibm.com/developerworks/cn/linux/1308_liuming_linuxtime3/
  2. 游戲服務器定時任務大家是通過什么方式實現的 韋易笑的回答。地址:https://www.zhihu.com/question/32251997


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM