Go使用TimerController解決timer過多的問題
背景
- 在Go里面我們實現(xiàn)超時需要起一個goroutine才能實現(xiàn),但是當我有大量的任務需要做超時控制就需要起大量的goroutine,實際上是一種開銷和負擔!
- 有些時候需要注冊一些Timer也是有需要起大量的 goroutine才能實現(xiàn),比如我要異步定期刷新一個配置,異步的監(jiān)聽啥東西,此時簡單做法就是使用大量的 goroutine + timer/sleep實現(xiàn)!
解決思路
多路復用,實際上Go底層也是一種多路復用的思想去實現(xiàn)的timer,但是它是底層的timer,我們需要解決的問題就過多的timer問題!
我們的思路是實現(xiàn)一個 TimerController 可以幫助我們管理很多個timer,并且可以開銷做到最低!因此使用一個 小頂堆 + Timer調度器即可實現(xiàn)!
實現(xiàn)
小頂堆(最小堆)
使用Go自帶的 container/heap 實現(xiàn) 小頂堆
import (
"container/heap"
)
type HeapItem[T any] interface {
Less(HeapItem[T]) bool
GetValue() T
}
// 參考 IntHeap
type heapQueue[T any] []HeapItem[T]
func (h heapQueue[T]) Len() int { return len(h) }
func (h heapQueue[T]) Less(i, j int) bool { return h[i].Less(h[j]) }
func (h heapQueue[T]) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h *heapQueue[T]) Push(x any) {
// Push and Pop use pointer receivers because they modify the slice's length,
// not just its contents.
*h = append(*h, x.(HeapItem[T]))
}
func (h *heapQueue[T]) Pop() any {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
type HeapQueue[T any] struct {
queue heapQueue[T]
}
func (h *HeapQueue[T]) ptr() *heapQueue[T] {
return &h.queue
}
// NewHeapQueue 非并發(fā)安全
func NewHeapQueue[T any](items ...HeapItem[T]) *HeapQueue[T] {
queue := make(heapQueue[T], len(items))
for index, item := range items {
queue[index] = item
}
heap.Init(&queue)
return &HeapQueue[T]{queue: queue}
}
func (h *HeapQueue[T]) Push(item HeapItem[T]) {
heap.Push(h.ptr(), item)
}
func (h *HeapQueue[T]) Pop() (T, bool ) {
if h.ptr().Len() == 0 {
var Nil T
return Nil, false
}
return heap.Pop(h.ptr()).(HeapItem[T]).GetValue(), true
}
// Peek 方法用于返回堆頂元素而不移除它
func (h *HeapQueue[T]) Peek() (T, bool ) {
if h.ptr().Len() > 0 {
return h.queue[0].GetValue(), true
}
var Nil T
return Nil, false
}
func (h *HeapQueue[T]) Len() int {
return h.ptr().Len()
}
調度器
type Timer struct {
Timeout time.Time
Name string
NotifyFunc func()
}
func (t *Timer) GetCurTimeout() time.Duration {
return t.Timeout.Sub(time.Now())
}
// Notify todo support async notify
func (t *Timer) Notify() {
if t.NotifyFunc != nil {
t.NotifyFunc()
}
}
func (t *Timer) IsExpired() bool {
return t.Timeout.Before(time.Now())
}
func (t *Timer) Less(v HeapItem[*Timer]) bool {
return t.Timeout.Before(v.GetValue().Timeout)
}
func (t *Timer) GetValue() *Timer {
return t
}
type TimerController struct {
timers chan *Timer
minHeap *HeapQueue[*Timer]
closeOnce sync.Once
close chan struct{}
}
func (t *TimerController) AddTimer(timer *Timer) bool {
if timer == nil {
return false
}
select {
case <-t.close:
return false
default:
t.timers <- timer
return true
}
}
func (t *TimerController) Close() {
t.closeOnce.Do(func() { close(t.close) })
}
func NewTimerController(bufferSize int) *TimerController {
return &TimerController{
timers: make(chan *Timer, bufferSize),
minHeap: NewHeapQueue[*Timer](),
close: make(chan struct{}),
}
}
func (t *TimerController) Start() {
go t._start()
}
func (t *TimerController) _start() {
const defaultTimeout = time.Hour * 24
var (
curMinTimer *Timer
timeout = time.NewTimer(defaultTimeout)
)
for {
select {
case <-t.close:
close(t.timers)
timeout.Stop()
return
case timer := <-t.timers:
t.minHeap.Push(timer)
curMinTimer, _ = t.minHeap.Peek()
timeout.Reset(curMinTimer.GetCurTimeout())
//fmt.Printf("timeout.Reset-1 name: %s, timeout: %s\n", curMinTimer.Name, curMinTimer.GetCurTimeout())
case <-timeout.C:
if curMinTimer != nil {
curMinTimer.Notify()
curMinTimer = nil
t.minHeap.Pop()
}
curMinTimer, _ = t.minHeap.Peek()
if curMinTimer == nil {
timeout.Reset(defaultTimeout)
continue
}
timeout.Reset(curMinTimer.GetCurTimeout())
//fmt.Printf("timeout.Reset-2 name: %s, timeout: %s\n", curMinTimer.Name, curMinTimer.GetCurTimeout())
}
}
}
測試
func TestTimerController(t *testing.T) {
controller := NewTimerController(1024)
controller.Start()
defer controller.Close()
now := time.Now()
arrs := make([]string, 0)
NewTimer := func(num int) *Timer {
return &Timer{Timeout: now.Add(time.Duration(num) * time.Millisecond), Name: strconv.Itoa(num), NotifyFunc: func() {
arrs = append(arrs, strconv.Itoa(num))
}}
}
// 這里亂序的注冊了8個timer
controller.AddTimer(NewTimer(5))
controller.AddTimer(NewTimer(6))
controller.AddTimer(NewTimer(3))
controller.AddTimer(NewTimer(4))
controller.AddTimer(NewTimer(7))
controller.AddTimer(NewTimer(8))
controller.AddTimer(NewTimer(1))
controller.AddTimer(NewTimer(2))
time.Sleep(time.Second * 1)
t.Logf("%#v\n", arrs)
// 最終我們可以獲取到 順序執(zhí)行的!
assert.Equal(t, arrs, []string{"1", "2", "3", "4", "5", "6", "7", "8"})
}
func TestTimerController_Stable(t *testing.T) {
controller := NewTimerController(1024)
controller.Start()
defer controller.Close()
now := time.Now()
arrs := make(map[string]bool, 0)
NewTimer := func(num int, name string) *Timer {
return &Timer{Timeout: now.Add(time.Duration(num) * time.Millisecond), Name: name, NotifyFunc: func() {
arrs[name] = true
}}
}
// 我們重復注冊了相同實現(xiàn)執(zhí)行的 timer,那么預期是每次執(zhí)行的結果和注冊順序一致
controller.AddTimer(NewTimer(2, "1"))
controller.AddTimer(NewTimer(2, "2"))
controller.AddTimer(NewTimer(2, "3"))
controller.AddTimer(NewTimer(2, "4"))
controller.AddTimer(NewTimer(2, "5"))
time.Sleep(time.Second * 1)
t.Logf("%#v\n", arrs)
assert.Equal(t, arrs, map[string]bool{"1": true, "2": true, "3": true, "4": true, "5": true})
}
以上就是Go使用TimerController解決timer過多的問題的詳細內容,更多關于Go TimerController解決timer過多的資料請關注腳本之家其它相關文章!
相關文章
Go語言輕松實現(xiàn)郵件發(fā)送通知功能的完全指南
在現(xiàn)代 Web 應用中,郵件通知是一個不可或缺的功能,本文將深入解析一個基于 Go 語言 smtp 協(xié)議和 email 庫的郵件發(fā)送工具,需要的可以了解下2025-04-04
Golang中的archive/zip包的常用函數(shù)詳解
Golang 中的 archive/zip 包用于處理 ZIP 格式的壓縮文件,提供了一系列用于創(chuàng)建、讀取和解壓縮 ZIP 格式文件的函數(shù)和類型,下面小編就來和大家講解下常用函數(shù)吧2023-08-08
Golang使用Apache PLC4X連接modbus的示例代碼
Modbus是一種串行通信協(xié)議,是Modicon公司于1979年為使用可編程邏輯控制器(PLC)通信而發(fā)表,這篇文章主要介紹了Golang使用Apache PLC4X連接modbus的示例代碼,需要的朋友可以參考下2024-07-07
GO語言中創(chuàng)建切片的三種實現(xiàn)方式
這篇文章主要介紹了GO語言中創(chuàng)建切片的三種實現(xiàn)方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-09-09

