fantasticbin/go-study
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化延时队列及队列集合的逻辑

Browse Source
This commit is contained in:
fantasticbin 2024-12-09 22:25:50 +08:00
parent 95c27f6e96
commit 800c590a3a
3 changed files with 44 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
lock_free/delay_queue.go
View File

@ -6,26 +6,26 @@ import (
"time" "time"
) )
type DelayLkQueue[TValue any, TKey comparable] struct { type DelayLkQueue[TKey comparable, TValue any] struct {
timers map[TKey]*time.Timer timers map[TKey]*time.Timer
m sync.Mutex m sync.Mutex
LkQueue[TValue] LkQueue[TValue]
} }
// NewDelayLkQueue 创建延迟无锁队列 // NewDelayLkQueue 创建延迟无锁队列
func NewDelayLkQueue[TValue any, TKey comparable]() *DelayLkQueue[TValue, TKey] { func NewDelayLkQueue[TKey comparable, TValue any]() *DelayLkQueue[TKey, TValue] {
return &DelayLkQueue[TValue, TKey]{make(map[TKey]*time.Timer), sync.Mutex{}, *NewLkQueue[TValue]()} return &DelayLkQueue[TKey, TValue]{make(map[TKey]*time.Timer), sync.Mutex{}, *NewLkQueue[TValue]()}
} }
// DelayEnqueue 延迟入队 // DelayEnqueue 延迟入队
func (q *DelayLkQueue[TValue, TKey]) DelayEnqueue(value TValue, duration time.Duration) { func (q *DelayLkQueue[TKey, TValue]) DelayEnqueue(value TValue, duration time.Duration) {
time.AfterFunc(duration, func() { time.AfterFunc(duration, func() {
q.Enqueue(value) q.Enqueue(value)
}) })
} }
// CancellableDelayEnqueue 可取消的延迟入队 // CancellableDelayEnqueue 可取消的延迟入队
func (q *DelayLkQueue[TValue, TKey]) CancellableDelayEnqueue(key TKey, value TValue, duration time.Duration) { func (q *DelayLkQueue[TKey, TValue]) CancellableDelayEnqueue(key TKey, value TValue, duration time.Duration) {
if timer, ok := q.timers[key]; ok { if timer, ok := q.timers[key]; ok {
timer.Stop() timer.Stop()
} }
@ -38,7 +38,7 @@ func (q *DelayLkQueue[TValue, TKey]) CancellableDelayEnqueue(key TKey, value TVa
} }
// CancelDelayEnqueue 取消延迟入队 // CancelDelayEnqueue 取消延迟入队
func (q *DelayLkQueue[TValue, TKey]) CancelDelayEnqueue(key TKey) { func (q *DelayLkQueue[TKey, TValue]) CancelDelayEnqueue(key TKey) {
q.m.Lock() q.m.Lock()
defer q.m.Unlock() defer q.m.Unlock()
if timer, ok := q.timers[key]; ok { if timer, ok := q.timers[key]; ok {
@ -48,7 +48,7 @@ func (q *DelayLkQueue[TValue, TKey]) CancelDelayEnqueue(key TKey) {
} }
// ContinuousDequeue 持续监听出队 // ContinuousDequeue 持续监听出队
func (q *DelayLkQueue[TValue, TKey]) ContinuousDequeue() iter.Seq[TValue] { func (q *DelayLkQueue[TKey, TValue]) ContinuousDequeue() iter.Seq[TValue] {
return func(yield func(TValue) bool) { return func(yield func(TValue) bool) {
for { for {
if value, ok := q.Dequeue(); ok { if value, ok := q.Dequeue(); ok {
@ -63,14 +63,14 @@ func (q *DelayLkQueue[TValue, TKey]) ContinuousDequeue() iter.Seq[TValue] {
} }
// ContinuousDequeueExecute 持续监听出队执行函数 // ContinuousDequeueExecute 持续监听出队执行函数
func (q *DelayLkQueue[TValue, TKey]) ContinuousDequeueExecute(fn func(TValue)) { func (q *DelayLkQueue[TKey, TValue]) ContinuousDequeueExecute(fn func(TValue)) {
for value := range q.ContinuousDequeue() { for value := range q.ContinuousDequeue() {
fn(value) fn(value)
} }
} }
// ContinuousDequeueNotify 持续监听出队通知 // ContinuousDequeueNotify 持续监听出队通知
func (q *DelayLkQueue[TValue, TKey]) ContinuousDequeueNotify(chs ...chan TValue) { func (q *DelayLkQueue[TKey, TValue]) ContinuousDequeueNotify(chs ...chan TValue) {
q.ContinuousDequeueExecute(func(value TValue) { q.ContinuousDequeueExecute(func(value TValue) {
for _, ch := range chs { for _, ch := range chs {
ch <- value ch <- value

93
lock_free/queues.go
View File

@ -7,106 +7,81 @@ import (
) )
// Queues 队列集合 // Queues 队列集合
type Queues[TValue any, TKey, TRoute comparable] struct { type Queues[TKey comparable, TValue, TRoute any] struct {
queues map[TRoute]*DelayLkQueue[TValue, TKey] queues sync.Map // 读多写少的场景，适合用 sync.Map 路由队列
m sync.RWMutex
} }
// NewQueues 创建队列集合 // NewQueues 创建队列集合
func NewQueues[TValue any, TKey, TRoute comparable]() *Queues[TValue, TKey, TRoute] { func NewQueues[TKey comparable, TValue, TRoute any]() *Queues[TKey, TValue, TRoute] {
return &Queues[TValue, TKey, TRoute]{make(map[TRoute]*DelayLkQueue[TValue, TKey]), sync.RWMutex{}} return &Queues[TKey, TValue, TRoute]{}
} }
// Enqueue 入队 // Enqueue 入队
func (q *Queues[TValue, TKey, TRoute]) Enqueue(route TRoute, value TValue) { func (q *Queues[TKey, TValue, TRoute]) Enqueue(route TRoute, value TValue) {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
if queue, ok := q.queues[route]; ok { queue.(*DelayLkQueue[TKey, TValue]).Enqueue(value)
q.m.RUnlock()
queue.Enqueue(value)
} else { } else {
q.m.RUnlock() queue := NewDelayLkQueue[TKey, TValue]()
queue := NewDelayLkQueue[TValue, TKey]() q.queues.Store(route, queue)
q.m.Lock()
q.queues[route] = queue
q.m.Unlock()
queue.Enqueue(value) queue.Enqueue(value)
} }
} }
// Dequeue 出队 // Dequeue 出队
func (q *Queues[TValue, TKey, TRoute]) Dequeue(route TRoute) (value TValue, ok bool) { func (q *Queues[TKey, TValue, TRoute]) Dequeue(route TRoute) (value TValue, ok bool) {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
defer q.m.RUnlock() return queue.(*DelayLkQueue[TKey, TValue]).Dequeue()
if queue, ok := q.queues[route]; ok {
return queue.Dequeue()
} }
return value, false return value, false
} }
// DelayEnqueue 延迟入队 // DelayEnqueue 延迟入队
func (q *Queues[TValue, TKey, TRoute]) DelayEnqueue(route TRoute, value TValue, duration time.Duration) { func (q *Queues[TKey, TValue, TRoute]) DelayEnqueue(route TRoute, value TValue, duration time.Duration) {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
if queue, ok := q.queues[route]; ok { queue.(*DelayLkQueue[TKey, TValue]).DelayEnqueue(value, duration)
q.m.RUnlock()
queue.DelayEnqueue(value, duration)
} else { } else {
q.m.RUnlock() queue := NewDelayLkQueue[TKey, TValue]()
queue := NewDelayLkQueue[TValue, TKey]() q.queues.Store(route, queue)
q.m.Lock()
q.queues[route] = queue
q.m.Unlock()
queue.DelayEnqueue(value, duration) queue.DelayEnqueue(value, duration)
} }
} }
// CancellableDelayEnqueue 可取消的延迟入队 // CancellableDelayEnqueue 可取消的延迟入队
func (q *Queues[TValue, TKey, TRoute]) CancellableDelayEnqueue(route TRoute, key TKey, value TValue, duration time.Duration) { func (q *Queues[TKey, TValue, TRoute]) CancellableDelayEnqueue(route TRoute, key TKey, value TValue, duration time.Duration) {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
defer q.m.RUnlock() queue.(*DelayLkQueue[TKey, TValue]).CancellableDelayEnqueue(key, value, duration)
if queue, ok := q.queues[route]; ok {
queue.CancellableDelayEnqueue(key, value, duration)
} else { } else {
queue := NewDelayLkQueue[TValue, TKey]() queue := NewDelayLkQueue[TKey, TValue]()
q.m.Lock() q.queues.Store(route, queue)
q.queues[route] = queue
q.m.Unlock()
queue.CancellableDelayEnqueue(key, value, duration) queue.CancellableDelayEnqueue(key, value, duration)
} }
} }
// CancelDelayEnqueue 取消延迟入队 // CancelDelayEnqueue 取消延迟入队
func (q *Queues[TValue, TKey, TRoute]) CancelDelayEnqueue(route TRoute, key TKey) { func (q *Queues[TKey, TValue, TRoute]) CancelDelayEnqueue(route TRoute, key TKey) {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
defer q.m.RUnlock() queue.(*DelayLkQueue[TKey, TValue]).CancelDelayEnqueue(key)
if queue, ok := q.queues[route]; ok {
queue.CancelDelayEnqueue(key)
} }
} }
// ContinuousDequeue 持续监听出队 // ContinuousDequeue 持续监听出队
func (q *Queues[TValue, TKey, TRoute]) ContinuousDequeue(route TRoute) iter.Seq[TValue] { func (q *Queues[TKey, TValue, TRoute]) ContinuousDequeue(route TRoute) iter.Seq[TValue] {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
defer q.m.RUnlock() return queue.(*DelayLkQueue[TKey, TValue]).ContinuousDequeue()
if queue, ok := q.queues[route]; ok {
return queue.ContinuousDequeue()
} }
return nil return nil
} }
// ContinuousDequeueExecute 持续监听出队执行函数 // ContinuousDequeueExecute 持续监听出队执行函数
func (q *Queues[TValue, TKey, TRoute]) ContinuousDequeueExecute(route TRoute, fn func(TValue)) { func (q *Queues[TKey, TValue, TRoute]) ContinuousDequeueExecute(route TRoute, fn func(TValue)) {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
defer q.m.RUnlock() queue.(*DelayLkQueue[TKey, TValue]).ContinuousDequeueExecute(fn)
if queue, ok := q.queues[route]; ok {
queue.ContinuousDequeueExecute(fn)
} }
} }
// ContinuousDequeueNotify 持续监听出队通知 // ContinuousDequeueNotify 持续监听出队通知
func (q *Queues[TValue, TKey, TRoute]) ContinuousDequeueNotify(route TRoute, chs ...chan TValue) { func (q *Queues[TKey, TValue, TRoute]) ContinuousDequeueNotify(route TRoute, chs ...chan TValue) {
q.m.RLock() if queue, ok := q.queues.Load(route); ok {
defer q.m.RUnlock() queue.(*DelayLkQueue[TKey, TValue]).ContinuousDequeueNotify(chs...)
if queue, ok := q.queues[route]; ok {
queue.ContinuousDequeueNotify(chs...)
} }
} }

2
lock_free/queues_test.go
View File

@ -14,7 +14,7 @@ func TestQueues(t *testing.T) {
{3, time.Second * 3}, {3, time.Second * 3},
} }
route := "test" route := "test"
q := NewQueues[int, struct{}, string]() q := NewQueues[struct{}, int, string]()
for _, c := range cases { for _, c := range cases {
q.DelayEnqueue(route, c.value, c.duration) q.DelayEnqueue(route, c.value, c.duration)