delay

Documentation

Overview

Package delay ...

Description : delay ...

Author : go_developer@163.com<白茶清欢>

Date : 2022-07-06 18:06

Package delay ...

Description : delay ...

Author : go_developer@163.com<白茶清欢>

Date : 2022-07-06 17:55

Package delay ...

Description : event ...

Author : go_developer@163.com<白茶清欢>

Date : 2025-08-22 18:55

Package delay ...

Description : 基于redis实现的延迟事件队列

Author : go_developer@163.com<白茶清欢>

Date : 2025-08-29 09:11

Package event ...

Description : 基于redis实现事件生产 + 消费

Author : go_developer@163.com<白茶清欢>

Date : 2025-08-22 16:34

Index

• func NewRedisConsumer(ctx context.Context, queueName string, queueCnt int, redisFlag string, ...) abstract.IConsumer
• func NewRedisDelayEvent(redisFlag string, queueName string, pullTimeInterval int64, ...) abstract.IDelayQueue
• func NewRedisProducer(queueName string, queueCnt int, redisFlag string, ...) abstract.IProducer
• type ConsumerConfig
• type IConsumer
• type IHandler
• type IProduce
• type ProduceData
• type Queue
  • func (q *Queue) Err() error
• type ZRangeData

Functions

func NewRedisConsumer

func NewRedisConsumer(
	ctx context.Context,
	queueName string, queueCnt int, redisFlag string,
	consumerHandler abstract.IConsumerHandler,
) abstract.IConsumer

NewRedisConsumer 启动一个消费者实例

func NewRedisDelayEvent

func NewRedisDelayEvent(redisFlag string, queueName string, pullTimeInterval int64, distributeProducer abstract.IProducer) abstract.IDelayQueue

NewRedisDelayEvent 获取延迟队列实例 参数说明:

• redisFlag: 使用那个redis实例
• queueName: 延迟队列名称
• pullTimeInterval: 延迟队列扫描的时间间隔, 单位秒
• distributeProducer: 消息生产者示例, 扫描到的数据通过此实例向二级任务队列分发

func NewRedisProducer

func NewRedisProducer(
	queueName string, queueCnt int, redisFlag string,
	producerHandler abstract.IProducerHandler,
) abstract.IProducer

NewRedisProducer 获取基于redis的生产者实例

Types

type ConsumerConfig

type ConsumerConfig struct {
	QueueName   string // 队列名称
	SonQueueCnt int    // 二级队列数量
	SonQueName  string // 二级队列名称
	HashKey     string // hash消息写到哪个二级队列的key, 若不配置或者key不存在, 使用 ProduceData.MsgID
}

ConsumerConfig 消费者配置

Author : go_developer@163.com<白茶清欢>

Date : 14:12 2022/7/8

type IConsumer

type IConsumer interface {
	// Consume 消费数据
	Consume(ctx context.Context) ([]*ProduceData, error)
	// ConsumeWithHandler 消费数据并使用handler处理
	ConsumeWithHandler(ctx context.Context, handler IHandler) error
}

IConsumer 消费者接口约束

Author : go_developer@163.com<白茶清欢>

Date : 10:19 2022/7/7

type IHandler

type IHandler interface {
	// Handle 处理消费到的数据
	Handle(queData []*ProduceData) error
}

IHandler 消息的处理

Author : go_developer@163.com<白茶清欢>

Date : 10:26 2022/7/7

type IProduce

type IProduce interface {
	// Produce 生产数据
	Produce(ctx context.Context, data ...*Queue) error
}

IProduce 生产者约束

Author : go_developer@163.com<白茶清欢>

Date : 18:06 2022/7/6

type ProduceData

type ProduceData struct {
	MsgID     string `json:"msg_id"`    // 消息ID
	Timestamp int64  `json:"timestamp"` // 消息生成时间: ms
	Host      string `json:"host"`      // 生产消息的机器IP
	Data      any    `json:"data"`      // 传入的业务数据
}

ProduceData 生产的数据结构

Author : go_developer@163.com<白茶清欢>

Date : 18:56 2022/7/6

type Queue

type Queue struct {
	Name      string `json:"name"`       // 队列名称
	DelayTime int64  `json:"delay_time"` // 延迟执行时间
	Data      any    `json:"data"`       // 入队数据
	// contains filtered or unexported fields
}

Queue 队列数据

Author : go_developer@163.com<白茶清欢>

Date : 17:56 2022/7/6

func (*Queue) Err

func (q *Queue) Err() error

Err 获取入队异常

Author : go_developer@163.com<白茶清欢>

Date : 18:37 2022/7/6

type ZRangeData

type ZRangeData struct {
	Score int64        // 数据分值
	Data  *ProduceData // 实际业务数据
}

ZRangeData zRange读取的数据

Author : go_developer@163.com<白茶清欢>

Date : 14:24 2022/7/8