go-study/lock_free/delay_queue.go

package lock_free

import (
	"iter"
	"time"
)

type DelayLkQueue[T any] struct {
	LkQueue[T]
}

// NewDelayLkQueue 创建延迟无锁队列
func NewDelayLkQueue[T any]() *DelayLkQueue[T] {
	return &DelayLkQueue[T]{*NewLkQueue[T]()}
}

// DelayEnqueue 延迟入队
func (q *DelayLkQueue[T]) DelayEnqueue(value T, duration time.Duration) {
	time.AfterFunc(duration, func() {
		q.Enqueue(value)
	})
}

// ContinuousDequeue 持续监听出队
func (q *DelayLkQueue[T]) ContinuousDequeue() iter.Seq[T] {
	return func(yield func(T) bool) {
		for {
			if value, ok := q.Dequeue(); ok {
				if !yield(value) {
					return
				}
			} else {
				time.Sleep(time.Millisecond) // 队列为空，休眠1毫秒
			}
		}
	}
}

// ContinuousDequeueExecute 持续监听出队执行函数
func (q *DelayLkQueue[T]) ContinuousDequeueExecute(fn func(T)) {
	for value := range q.ContinuousDequeue() {
		fn(value)
	}
}

// ContinuousDequeueNotify 持续监听出队通知
func (q *DelayLkQueue[T]) ContinuousDequeueNotify(chs ...chan T) {
	q.ContinuousDequeueExecute(func(value T) {
		for _, ch := range chs {
			ch <- value
		}
	})
}