From f4f9e38b982eff6f0a09c941111e8be873b8f80e Mon Sep 17 00:00:00 2001 From: tangmingyou <234767776@qq.com> Date: Fri, 24 Mar 2023 23:28:53 +0800 Subject: [PATCH] delay queue bugfix --- go.mod | 6 +++--- internal/service/event/auto_betting.go | 2 +- tool/collect/delay_queue.go | 10 ++++++---- 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/go.mod b/go.mod index de0e26e..c50e1e3 100644 --- a/go.mod +++ b/go.mod @@ -5,17 +5,19 @@ go 1.19 require ( github.com/BurntSushi/toml v1.2.1 github.com/dchest/captcha v1.0.0 + github.com/emirpasic/gods v1.18.1 github.com/gin-contrib/sessions v0.0.5 github.com/gin-gonic/gin v1.8.2 github.com/golang/protobuf v1.5.2 github.com/gorilla/websocket v1.5.0 + github.com/o1egl/govatar v0.4.1 + github.com/patrickmn/go-cache v2.1.0+incompatible google.golang.org/protobuf v1.28.1 gorm.io/driver/sqlite v1.4.4 gorm.io/gorm v1.24.3 ) require ( - github.com/emirpasic/gods v1.18.1 // indirect github.com/gin-contrib/sse v0.1.0 // indirect github.com/go-playground/locales v0.14.0 // indirect github.com/go-playground/universal-translator v0.18.0 // indirect @@ -32,8 +34,6 @@ require ( github.com/mattn/go-sqlite3 v2.0.3+incompatible // indirect github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect github.com/modern-go/reflect2 v1.0.2 // indirect - github.com/o1egl/govatar v0.4.1 // indirect - github.com/patrickmn/go-cache v2.1.0+incompatible // indirect github.com/pelletier/go-toml/v2 v2.0.6 // indirect github.com/ugorji/go/codec v1.2.7 // indirect golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3 // indirect diff --git a/internal/service/event/auto_betting.go b/internal/service/event/auto_betting.go index e04c836..8880639 100644 --- a/internal/service/event/auto_betting.go +++ b/internal/service/event/auto_betting.go @@ -15,7 +15,7 @@ var AutoBettingDelayQueue *collect.DelayQueue[int64] var HandleReqBetting func(player *game.Player, msg *api.ReqBetting) (proto.Message, error) func init() { - AutoBettingDelayQueue = collect.NewDelayQueue(handleAutoBetting) + AutoBettingDelayQueue = collect.NewDelayQueue(time.Second, handleAutoBetting) } func handleAutoBetting(accountId int64, _ time.Time) { diff --git a/tool/collect/delay_queue.go b/tool/collect/delay_queue.go index dbd0a00..47bcb49 100644 --- a/tool/collect/delay_queue.go +++ b/tool/collect/delay_queue.go @@ -14,16 +14,18 @@ type DelayQueue[T any] struct { taskQueueMap *treemap.Map lock *sync.RWMutex offsetMs int64 + checkInterval time.Duration handler func(data T, now time.Time) supplying *atomic.Bool consumeChannel chan []T } -func NewDelayQueue[T any](handler func(data T, now time.Time)) *DelayQueue[T] { +func NewDelayQueue[T any](checkInterval time.Duration, handler func(data T, now time.Time)) *DelayQueue[T] { q := &DelayQueue[T]{} q.taskQueueMap = treemap.NewWithIntComparator() q.lock = &sync.RWMutex{} q.offsetMs = time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli() + q.checkInterval = checkInterval q.handler = handler q.supplying = &atomic.Bool{} q.consumeChannel = make(chan []T, 32) @@ -62,7 +64,7 @@ func (q *DelayQueue[T]) supplier() { } go func() { for { - now := <-time.After(time.Millisecond * 100) + now := <-time.After(q.checkInterval) _, queue := q.next(now.UnixMilli()) if IsEmptySlice(queue) { continue @@ -98,6 +100,8 @@ func (q *DelayQueue[T]) next(nowMs int64) (int64, []T) { // Add 添加任务 func (q *DelayQueue[T]) Add(after time.Duration, data T) int64 { ms := time.Now().UnixMilli() + after.Milliseconds() + // 通知生产者goroutine轮训数据 + defer q.supplier() q.lock.Lock() defer q.lock.Unlock() val, found := q.taskQueueMap.Get(int(ms)) @@ -107,8 +111,6 @@ func (q *DelayQueue[T]) Add(after time.Duration, data T) int64 { } values := val.([]T) q.taskQueueMap.Put(int(ms), append(values, data)) - // 通知生产者goroutine轮训数据 - q.supplier() return q.genKey(ms, len(values)) }