Go語言中時間輪的實現

轉載請聲明出處哦~,本篇文章發佈於luozhiyun的博客://www.luozhiyun.com/archives/444

最近在工作中有一個需求,簡單來說就是在短時間內會創建上百萬個定時任務,創建的時候會將對應的金額相加,防止超售,需要過半個小時再去核對數據,如果數據對不上就需要將加上的金額再減回去。

這個需求如果用Go內置的Timer來做的話性能比較低下,因為Timer是使用最小堆來實現的,創建和刪除的時間複雜度都為 O(log n)。如果使用時間輪的話則是O(1)性能會好很多。

對於時間輪來說,我以前寫過一篇java版的時間輪算法分析://www.luozhiyun.com/archives/59,這次來看看Go語言的時間輪實現,順便大家有興趣的也可以對比一下兩者的區別,以及我寫文章的水平和一年多前有沒有提升,哈哈哈。

時間輪的運用其實是非常的廣泛的,在 Netty、Akka、Quartz、ZooKeeper、Kafka 等組件中都存在時間輪的蹤影。下面用Go實現的時間輪是以Kafka的代碼為原型來實現的,完整代碼://github.com/devYun/timingwheel。

介紹

簡單時間輪

在時間輪中存儲任務的是一個環形隊列,底層採用數組實現,數組中的每個元素可以存放一個定時任務列表。定時任務列表是一個環形的雙向鏈表,鏈表中的每一項表示的都是定時任務項,其中封裝了真正的定時任務。

時間輪由多個時間格組成,每個時間格代表當前時間輪的基本時間跨度(tickMs)。時間輪的時間格個數是固定的,可用 wheelSize 來表示,那麼整個時間輪的總體時間跨度(interval)可以通過公式 tickMs×wheelSize 計算得出。

時間輪還有一個錶盤指針(currentTime),用來表示時間輪當前所處的時間,currentTime 是 tickMs 的整數倍。currentTime指向的地方是表示到期的時間格,表示需要處理的時間格所對應的鏈表中的所有任務。

如下圖是一個tickMs為1s,wheelSize等於10的時間輪,每一格裏面放的是一個定時任務鏈表,鏈表裏面存有真正的任務項:

taskList

初始情況下錶盤指針 currentTime 指向時間格0,若時間輪的 tickMs 為 1ms 且 wheelSize 等於10,那麼interval則等於10s。如下圖此時有一個定時為2s的任務插進來會存放到時間格為2的任務鏈表中,用紅色標記。隨着時間的不斷推移,指針 currentTime 不斷向前推進,如果過了2s,那麼 currentTime 會指向時間格2的位置,會將此時間格的任務鏈表獲取出來處理。

timewheel

如果當前的指針 currentTime 指向的是2,此時如果插入一個9s的任務進來,那麼新來的任務會服用原來的時間格鏈表,會存放到時間格1中

timewheelAdd9S

這裡所講的時間輪都是簡單時間輪,只有一層,總體時間範圍在 currentTime 和 currentTime+interval 之間。如果現在有一個15s的定時任務是需要重新開啟一個時間輪,設置一個時間跨度至少為15s的時間輪才夠用。但是這樣擴充是沒有底線的,如果需要一個1萬秒的時間輪,那麼就需要一個這麼大的數組去存放,不僅佔用很大的內存空間,而且也會因為需要遍歷這麼大的數組從而拉低效率。

因此引入了層級時間輪的概念。

層級時間輪

如圖是一個兩層的時間輪,第二層時間輪也是由10個時間格組成,每個時間格的跨度是10s。第二層的時間輪的 tickMs 為第一層時間輪的 interval,即10s。每一層時間輪的 wheelSize 是固定的,都是10,那麼第二層的時間輪的總體時間跨度 interval 為100s。

圖中展示了每個時間格對應的過期時間範圍, 我們可以清晰地看到, 第二層時間輪的第0個時間格的過期時間範圍是 [0,9]。也就是說, 第二層時間輪的一個時間格就可以表示第一層時間輪的所有(10個)時間格;

如果向該時間輪中添加一個15s的任務,那麼當第一層時間輪容納不下時,進入第二層時間輪,並插入到過期時間為[10,19]的時間格中。

timewheellevel2

隨着時間的流逝,當原本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 中到期的任務;

結構體

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是上層時間輪的引用。

type bucket struct {
	// 任務的過期時間
	expiration int64

	mu     sync.Mutex
	// 相同過期時間的任務隊列
	timers *list.List
}

bucket裏面實際上封裝的是時間格裏面的任務隊列,裏面放入的是相同過期時間的任務,到期後會將隊列timers拿出來進行處理。這裡有個有意思的地方是由於會有多個線程並發的訪問bucket,所以需要用到原子類來獲取int64位的值,為了保證32位系統上面讀取64位數據的一致性,需要進行64位對齊。具體的可以看這篇://www.luozhiyun.com/archives/429,講的是對內存對齊的思考。

type Timer struct {
  // 到期時間
	expiration int64 // in milliseconds
  // 要被執行的具體任務
	task       func()
	// Timer所在bucket的指針
	b unsafe.Pointer // type: *bucket
	// bucket列表中對應的元素
	element *list.Element
}

Timer是時間輪的最小執行單元,是定時任務的封裝,到期後會調用task來執行任務。

Group 37

初始化時間輪

例如現在初始化一個tick是1s,wheelSize是10的時間輪:

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方法:

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方法執行。

func (tw *TimingWheel) advanceClock(expiration int64) {
	currentTime := atomic.LoadInt64(&tw.currentTime)
	// 過期時間大於等於(當前時間+tick)
	if expiration >= currentTime+tw.tick {
		// 將currentTime設置為expiration,從而推進currentTime
		currentTime = truncate(expiration, tw.tick)
		atomic.StoreInt64(&tw.currentTime, currentTime)

		// Try to advance the clock of the overflow wheel if present
		// 如果有上層時間輪,那麼遞歸調用上層時間輪的引用
		overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
		if overflowWheel != nil {
			(*TimingWheel)(overflowWheel).advanceClock(currentTime)
		}
	}
}

advanceClock方法會根據到期時間來從新設置currentTime,從而推進時間輪前進。

func (b *bucket) Flush(reinsert func(*Timer)) {
	var ts []*Timer

	b.mu.Lock()
	// 循環獲取bucket隊列節點
	for e := b.timers.Front(); e != nil; {
		next := e.Next()

		t := e.Value.(*Timer)
		// 將頭節點移除bucket隊列
		b.remove(t)
		ts = append(ts, t)

		e = next
	}
	b.mu.Unlock()

	b.SetExpiration(-1) // TODO: Improve the coordination with b.Add()

	for _, t := range ts {
		reinsert(t)
	}
}

Flush方法會根據bucket裏面timers列表進行遍歷插入到ts數組中,然後調用reinsert方法,這裡是調用的addOrRun方法。

func (tw *TimingWheel) addOrRun(t *Timer) {
	// 如果已經過期,那麼直接執行
	if !tw.add(t) { 
		// 異步執行定時任務
		go t.task()
	}
}

addOrRun會調用add方法檢查傳入的定時任務Timer是否已經到期,如果到期那麼異步調用task方法直接執行。add方法我們下面會接着分析。

整個start執行流程如圖:

timewheel_start

  1. start方法回啟動一個goroutines調用poll來處理DelayQueue中到期的數據,並將數據放入到管道C中;
  2. start方法啟動第二個goroutines方法會循環獲取DelayQueue中管道C的數據,管道C中實際上存放的是一個bucket,然後遍歷bucket的timers列表,如果任務已經到期,那麼異步執行,沒有到期則重新放入到DelayQueue中。

add task

func main() {
	tw := timingwheel.NewTimingWheel(time.Second, 10)
	tw.Start() 
	// 添加任務
	tw.AfterFunc(time.Second*15, func() {
		fmt.Println("The timer fires")
		exitC <- time.Now().UTC()
	})
}

我們通過AfterFunc方法添加一個15s的定時任務,如果到期了,那麼執行傳入的函數。

func (tw *TimingWheel) AfterFunc(d time.Duration, f func()) *Timer {
	t := &Timer{
		expiration: timeToMs(time.Now().UTC().Add(d)),
		task:       f,
	}
	tw.addOrRun(t)
	return t
}

AfterFunc方法回根據傳入的任務到期時間,以及到期需要執行的函數封裝成Timer,調用addOrRun方法。addOrRun方法我們上面已經看過了,會根據到期時間來決定是否需要執行定時任務。

下面我們來看一下add方法:

func (tw *TimingWheel) add(t *Timer) bool {
	currentTime := atomic.LoadInt64(&tw.currentTime)
	// 已經過期
	if t.expiration < currentTime+tw.tick {
		// Already expired
		return false
	// 	到期時間在第一層環內
	} else if t.expiration < currentTime+tw.interval {
		// Put it into its own bucket
		// 獲取時間輪的位置
		virtualID := t.expiration / tw.tick
		b := tw.buckets[virtualID%tw.wheelSize]
		// 將任務放入到bucket隊列中
		b.Add(t) 
		// 如果是相同的時間,那麼返回false,防止被多次插入到隊列中
		if b.SetExpiration(virtualID * tw.tick) { 
			// 將該bucket加入到延遲隊列中
			tw.queue.Offer(b, b.Expiration())
		}

		return true
	} else {
		// Out of the interval. Put it into the overflow wheel
		// 如果放入的到期時間超過第一層時間輪,那麼放到上一層中去
		overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
		if overflowWheel == nil {
			atomic.CompareAndSwapPointer(
				&tw.overflowWheel,
				nil,
				// 需要注意的是,這裡tick變成了interval
				unsafe.Pointer(newTimingWheel(
					tw.interval,
					tw.wheelSize,
					currentTime,
					tw.queue,
				)),
			)
			overflowWheel = atomic.LoadPointer(&tw.overflowWheel)
		}
		// 往上遞歸
		return (*TimingWheel)(overflowWheel).add(t)
	}
}

add方法根據到期時間來分成了三部分,第一部分是小於當前時間+tick,表示已經到期,那麼返回false執行任務即可;

第二部分的判斷會根據expiration是否小於時間輪的跨度,如果小於的話表示該定時任務可以放入到當前時間輪中,通過取模找到buckets對應的時間格並放入到bucket隊列中,SetExpiration方法會根據傳入的參數來判斷是否已經執行過延遲隊列的Offer方法,防止重複插入;

第三部分表示該定時任務的時間跨度超過了當前時間輪,需要升級到上一層的時間輪中。需要注意的是,上一層的時間輪的tick是當前時間輪的interval,延遲隊列還是同一個,然後設置為指針overflowWheel,並調用add方法往上層遞歸。

到這裡時間輪已經講完了,不過還有需要注意的地方,我們在用上面的時間輪實現中,使用了DelayQueue加環形隊列的方式實現了時間輪。對定時任務項的插入和刪除操作而言,TimingWheel時間複雜度為 O(1),在DelayQueue中的隊列使用的是優先隊列,時間複雜度是O(log n),但是由於buckets列表實際上是非常小的,所以並不會影響性能。

Reference

//github.com/RussellLuo/timingwheel

//zhuanlan.zhihu.com/p/121483218

//github.com/apache/kafka/tree/3cdc78e6bb1f83973a14ce1550fe3874f7348b05/core/src/main/scala/kafka/utils/timer

Tags: