Go語言中時間輪的實現
阿新 • • 發佈：2021-02-13
> 轉載請宣告出處哦~，本篇文章釋出於luozhiyun的部落格：https://www.luozhiyun.com/archives/444
最近在工作中有一個需求，簡單來說就是在短時間內會建立上百萬個定時任務，建立的時候會將對應的金額相加，防止超售，需要過半個小時再去核對資料，如果資料對不上就需要將加上的金額再減回去。
這個需求如果用Go內建的Timer來做的話效能比較低下，因為Timer是使用最小堆來實現的，建立和刪除的時間複雜度都為 O(log n)。如果使用時間輪的話則是O(1)效能會好很多。
對於時間輪來說，我以前寫過一篇java版的時間輪演算法分析：https://www.luozhiyun.com/archives/59，這次來看看Go語言的時間輪實現，順便大家有興趣的也可以對比一下兩者的區別，以及我寫文章的水平和一年多前有沒有提升，哈哈哈。
時間輪的運用其實是非常的廣泛的，在 Netty、Akka、Quartz、ZooKeeper、Kafka 等元件中都存在時間輪的蹤影。下面用Go實現的時間輪是以Kafka的程式碼為原型來實現的，完整程式碼：https://github.com/devYun/timingwheel。
## 介紹
### 簡單時間輪
在時間輪中儲存任務的是一個環形佇列，底層採用陣列實現，陣列中的每個元素可以存放一個定時任務列表。定時任務列表是一個環形的雙向連結串列，連結串列中的每一項表示的都是定時任務項，其中封裝了真正的定時任務。
時間輪由多個時間格組成，每個時間格代表當前時間輪的基本時間跨度（tickMs）。時間輪的時間格個數是固定的，可用 wheelSize 來表示，那麼整個時間輪的總體時間跨度（interval）可以通過公式 tickMs×wheelSize 計算得出。
時間輪還有一個錶盤指標（currentTime），用來表示時間輪當前所處的時間，currentTime 是 tickMs 的整數倍。currentTime指向的地方是表示到期的時間格，表示需要處理的時間格所對應的連結串列中的所有任務。
如下圖是一個tickMs為1s，wheelSize等於10的時間輪，每一格里面放的是一個定時任務連結串列，連結串列裡面存有真正的任務項：
初始情況下表盤指標 currentTime 指向時間格0，若時間輪的 tickMs 為 1ms 且 wheelSize 等於10，那麼interval則等於10s。如下圖此時有一個定時為2s的任務插進來會存放到時間格為2的任務連結串列中，用紅色標記。隨著時間的不斷推移，指標 currentTime 不斷向前推進，如果過了2s，那麼 currentTime 會指向時間格2的位置，會將此時間格的任務連結串列獲取出來處理。
如果當前的指標 currentTime 指向的是2，此時如果插入一個9s的任務進來，那麼新來的任務會服用原來的時間格連結串列，會存放到時間格1中
這裡所講的時間輪都是簡單時間輪，只有一層，總體時間範圍在 currentTime 和 currentTime+interval 之間。如果現在有一個15s的定時任務是需要重新開啟一個時間輪，設定一個時間跨度至少為15s的時間輪才夠用。但是這樣擴充是沒有底線的，如果需要一個1萬秒的時間輪，那麼就需要一個這麼大的陣列去存放，不僅佔用很大的記憶體空間，而且也會因為需要遍歷這麼大的陣列從而拉低效率。
因此引入了層級時間輪的概念。
### 層級時間輪
如圖是一個兩層的時間輪，第二層時間輪也是由10個時間格組成，每個時間格的跨度是10s。第二層的時間輪的 tickMs 為第一層時間輪的 interval，即10s。每一層時間輪的 wheelSize 是固定的，都是10，那麼第二層的時間輪的總體時間跨度 interval 為100s。
圖中展示了每個時間格對應的過期時間範圍， 我們可以清晰地看到， 第二層時間輪的第0個時間格的過期時間範圍是 [0,9]。也就是說, 第二層時間輪的一個時間格就可以表示第一層時間輪的所有(10個)時間格；
如果向該時間輪中新增一個15s的任務，那麼當第一層時間輪容納不下時，進入第二層時間輪，並插入到過期時間為[10，19]的時間格中。
隨著時間的流逝，當原本15s的任務還剩下5s的時候，這裡就有一個時間輪降級的操作，此時第一層時間輪的總體時間跨度已足夠，此任務被新增到第一層時間輪到期時間為5的時間格中，之後再經歷5s後，此任務真正到期，最終執行相應的到期操作。
## 程式碼實現
因為我們這個Go語言版本的時間輪程式碼是仿照Kafka寫的，所以在具體實現時間輪 TimingWheel 時還有一些小細節：
* 時間輪的時間格中每個連結串列會有一個root節點用於簡化邊界條件。它是一個附加的連結串列節點，該節點作為第一個節點，它的值域中並不儲存任何東西，只是為了操作的方便而引入的；
* 除了第一層時間輪，其餘高層時間輪的起始時間（startMs）都設定為建立此層時間輪時前面第一輪的 currentTime。每一層的 currentTime 都必須是 tickMs 的整數倍，如果不滿足則會將 currentTime 修剪為 tickMs 的整數倍。修剪方法為：currentTime = startMs - (startMs % tickMs)；
* Kafka 中的定時器只需持有 TimingWheel 的第一層時間輪的引用，並不會直接持有其他高層的時間輪，但每一層時間輪都會有一個引用（overflowWheel）指向更高一層的應用；
* Kafka 中的定時器使用了 DelayQueue 來協助推進時間輪。在操作中會將每個使用到的時間格中每個連結串列都加入 DelayQueue，DelayQueue 會根據時間輪對應的過期時間 expiration 來排序，最短 expiration 的任務會被排在 DelayQueue 的隊頭，通過單獨執行緒來獲取 DelayQueue 中到期的任務；
### 結構體
```go
type TimingWheel struct {
// 時間跨度,單位是毫秒
tick int64 // in milliseconds
// 時間輪個數
wheelSize int64
// 總跨度
interval int64 // in milliseconds
// 當前指標指向時間
currentTime int64 // in milliseconds
// 時間格列表
buckets []*bucket
// 延遲佇列
queue *delayqueue.DelayQueue
// 上級的時間輪引用
overflowWheel unsafe.Pointer // type: *TimingWheel
exitC chan struct{}
waitGroup waitGroupWrapper
}
```
tick、wheelSize、interval、currentTime都比較好理解，buckets欄位代表的是時間格列表，queue是一個延遲佇列，所有的任務都是通過延遲佇列來進行觸發，overflowWheel是上層時間輪的引用。
```go
type bucket struct {
// 任務的過期時間
expiration int64
mu sync.Mutex
// 相同過期時間的任務佇列
timers *list.List
}
```
bucket裡面實際上封裝的是時間格里面的任務佇列，裡面放入的是相同過期時間的任務，到期後會將佇列timers拿出來進行處理。這裡有個有意思的地方是由於會有多個執行緒併發的訪問bucket，所以需要用到原子類來獲取int64位的值，為了保證32位系統上面讀取64位資料的一致性，需要進行64位對齊。具體的可以看這篇：https://www.luozhiyun.com/archives/429，講的是對記憶體對齊的思考。
```go
type Timer struct {
// 到期時間
expiration int64 // in milliseconds
// 要被執行的具體任務
task func()
// Timer所在bucket的指標
b unsafe.Pointer // type: *bucket
// bucket列表中對應的元素
element *list.Element
}
```
Timer是時間輪的最小執行單元，是定時任務的封裝，到期後會呼叫task來執行任務。
### 初始化時間輪
例如現在初始化一個tick是1s，wheelSize是10的時間輪：
```go
func main() {
tw := timingwheel.NewTimingWheel(time.Second, 10)
tw.Start()
}
func NewTimingWheel(tick time.Duration, wheelSize int64) *TimingWheel {
// 將傳入的tick轉化成毫秒
tickMs := int64(tick / time.Millisecond)
// 如果小於零，那麼panic
if tickMs <= 0 {
panic(errors.New("tick must be greater than or equal to 1ms"))
}
// 設定開始時間
startMs := timeToMs(time.Now().UTC())
// 初始化TimingWheel
return newTimingWheel(
tickMs,
wheelSize,
startMs,
delayqueue.New(int(wheelSize)),
)
}
func newTimingWheel(tickMs int64, wheelSize int64, startMs int64, queue *delayqueue.DelayQueue) *TimingWheel {
// 初始化buckets的大小
buckets := make([]*bucket, wheelSize)
for i := range buckets {
buckets[i] = newBucket()
}
// 例項化TimingWheel
return &TimingWheel{
tick: tickMs,
wheelSize: wheelSize,
// currentTime必須是tickMs的倍數，所以這裡使用truncate進行修剪
currentTime: truncate(startMs, tickMs),
interval: tickMs * wheelSize,
buckets: buckets,
queue: queue,
exitC: make(chan struct{}),
}
}
```
初始化十分簡單，大家可以看看上面的程式碼註釋即可。
### 啟動時間輪
下面我們看看start方法：
```go
func (tw *TimingWheel) Start() {
// Poll會執行一個無限迴圈，將到期的元素放入到queue的C管道中
tw.waitGroup.Wrap(func() {
tw.queue.Poll(tw.exitC, func() int64 {
return timeToMs(time.Now().UTC())
})
})
// 開啟無限迴圈獲取queue中C的資料
tw.waitGroup.Wrap(func() {
for {
select {
// 從佇列裡面出來的資料都是到期的bucket
case elem := <-tw.queue.C:
b := elem.(*bucket)
// 時間輪會將當前時間 currentTime 往前移動到 bucket的到期時間
tw.advanceClock(b.Expiration())
// 取出bucket佇列的資料，並呼叫addOrRun方法執行
b.Flush(tw.addOrRun)
case <-tw.exitC:
return
}
}
})
}
```
這裡使用了util封裝的一個Wrap方法，這個方法會起一個goroutines非同步執行傳入的函式，具體的可以到我上面給出的連結去看原始碼。
Start方法會啟動兩個goroutines。第一個goroutines用來呼叫延遲佇列的queue的Poll方法，這個方法會一直迴圈獲取佇列裡面的資料，然後將到期的資料放入到queue的C管道中；第二個goroutines會無限迴圈獲取queue中C的資料，如果C中有資料表示已經到期，那麼會先呼叫advanceClock方法將當前時間 currentTime 往前移動到 bucket的到期時間，然後再呼叫Flush方法取出bucket中的佇列，並呼叫addOrRun方法執行。
```go
func (tw *TimingWheel) advanceClock(expiration int64) {
currentTime := atomic.LoadInt64(&tw.currentTime)
// 過期時間大於等於（當前時間+tick）
if expiration >