service

Documentation

Index

• Variables
• type IQueue
  • func GetQueueByName(name string) IQueue[any]
  • func GetQueues() []IQueue[any]
• type MemoryQueue
  • func NewMemoryQueue[T any](name string, capacity int, consumeSpeed int) *MemoryQueue[T]
  • func (q *MemoryQueue[T]) Close()
  • func (q *MemoryQueue[T]) Dequeue(ctx context.Context) (T, error)
  • func (q *MemoryQueue[T]) Enqueue(ctx context.Context, item T) error
  • func (q *MemoryQueue[T]) GetConsumerCount() int
  • func (q *MemoryQueue[T]) GetConsumerRunning() bool
  • func (q *MemoryQueue[T]) GetStatus() interface{}
  • func (q *MemoryQueue[T]) RunConsume(ctx context.Context, message_processor func(T))
  • func (q *MemoryQueue[T]) SetConsumerCount(count int) int
  • func (q *MemoryQueue[T]) SetConsumerRunning(status bool) bool
  • func (q *MemoryQueue[T]) WaitUntilEmpty()

Variables

var GeniricQueue = NewMemoryQueue[any]("geniricQueue", 100, 1)

var IntQueue = NewMemoryQueue[any]("intQueue", 100, 1)

var StringQueue = NewMemoryQueue[any]("stringQueue", 100, 1)

Types

type IQueue

type IQueue[T any] interface {
	// SetStatus 设置队列运行状态
	SetConsumerRunning(status bool) bool
	// Status 返回队列全部状态
	GetStatus() interface{}
	// 根据key查询对应的值
	GetConsumerRunning() bool
	// SetStatus 设置队列运行状态
	SetConsumerCount(count int) int
	// Enqueue 将项目放入队列
	Enqueue(ctx context.Context, item T) error
	// Dequeue 从队列中取出项目
	Dequeue(ctx context.Context) (T, error)
	// 启动消费者
	RunConsume(ctx context.Context, message_processor func(T))

	// Close 关闭队列
	Close()
	// WaitUntilEmpty 等待队列为空
	WaitUntilEmpty()
}

func GetQueueByName

func GetQueueByName(name string) IQueue[any]

func GetQueues

func GetQueues() []IQueue[any]

type MemoryQueue

type MemoryQueue[T any] struct {
	Name            string // 队列的名称
	Capacity        int
	ConsumerRunning bool
	ConsumerCount   int
	ConsumeSpeed    int
	// contains filtered or unexported fields
}

func NewMemoryQueue

func NewMemoryQueue[T any](name string, capacity int, consumeSpeed int) *MemoryQueue[T]

func (*MemoryQueue[T]) Close

func (q *MemoryQueue[T]) Close()

func (*MemoryQueue[T]) Dequeue

func (q *MemoryQueue[T]) Dequeue(ctx context.Context) (T, error)

Dequeue 从队列中取出一个项目

func (*MemoryQueue[T]) Enqueue

func (q *MemoryQueue[T]) Enqueue(ctx context.Context, item T) error

func (*MemoryQueue[T]) GetConsumerCount

func (q *MemoryQueue[T]) GetConsumerCount() int

func (*MemoryQueue[T]) GetConsumerRunning

func (q *MemoryQueue[T]) GetConsumerRunning() bool

func (*MemoryQueue[T]) GetStatus

func (q *MemoryQueue[T]) GetStatus() interface{}

func (*MemoryQueue[T]) RunConsume

func (q *MemoryQueue[T]) RunConsume(ctx context.Context, message_processor func(T))

func (*MemoryQueue[T]) SetConsumerCount

func (q *MemoryQueue[T]) SetConsumerCount(count int) int

func (*MemoryQueue[T]) SetConsumerRunning

func (q *MemoryQueue[T]) SetConsumerRunning(status bool) bool

func (*MemoryQueue[T]) WaitUntilEmpty

func (q *MemoryQueue[T]) WaitUntilEmpty()