delay queue bugfix

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

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) {

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))
 }