redq

package module
v0.0.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 2, 2026 License: MIT Imports: 14 Imported by: 0

README

redq

Go Redis Delivery

中文文档

redq is a production-oriented delay queue for Go, built on Redis.

The name means Redis delay queue and also evokes re-dq: a focused take on delay queues with Redis-native building blocks.

redq stores delayed tasks in Redis ZSets, moves due tasks into Redis Streams, and consumes them through Redis Stream Consumer Groups. It is designed for services that need delayed execution, retry orchestration, dead-letter handling, Redis Cluster-safe keys, and predictable operational semantics.

Producer -> Redis ZSet -> Scheduler -> Redis Stream -> Consumer Group -> Handler

Highlights

  • Redis-backed delayed task scheduling.
  • Redis Cluster-safe key layout with one hash tag per queue shard.
  • Multi-shard queues for horizontal scale.
  • Scheduler shard leasing and renewal.
  • Redis Stream Consumer Group delivery.
  • Bounded consumer concurrency.
  • At-least-once delivery with explicit idempotency guidance.
  • Automatic retry with fixed or exponential backoff.
  • Dead-letter queue listing, single-message replay, and batch replay.
  • Pending message recovery through XAUTOCLAIM.
  • Payload storage separated from ZSet and Stream entries.
  • Payload cleanup on successful ack by default.
  • Stream entry deletion on successful ack by default.
  • Missing-payload isolation through DLQ tombstones.
  • Lua scripts that verify pending ownership before ack, retry, or DLQ transitions.
  • Unit, integration, race, and benchmark coverage.

Installation

go get github.com/pudonghot/redq

redq requires Go 1.24+ and Redis with ZSet, Stream, Consumer Group, Lua, and XAUTOCLAIM support.

Quick Start

Start a local Redis:

docker compose up -d redis

The local Docker Compose Redis password is:

123456

Create a queue, enqueue a task, run the scheduler, and consume due tasks:

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/pudonghot/redq"
)

type EmailHandler struct{}

func (EmailHandler) Handle(ctx context.Context, task *redq.Task) error {
	fmt.Printf("handle task=%s payload=%s\n", task.ID, string(task.Payload))
	return nil
}

func main() {
	ctx := context.Background()

	client, err := redq.NewRedisClient(redq.RedisConfig{
		Addrs:    []string{"localhost:6379"},
		Password: "123456",
	})
	if err != nil {
		panic(err)
	}
	defer client.Close()

	queue := redq.QueueConfig{
		Name:            "email",
		Shards:          16,
		MaxPayloadBytes: 256 * 1024,
		DefaultAttempts: 3,
	}

	producer, err := redq.NewProducer(client, redq.ProducerOptions{
		Namespace: "redq",
		Queue:     queue,
	})
	if err != nil {
		panic(err)
	}

	result, err := producer.Enqueue(ctx, redq.Task{
		Payload:        []byte(`{"user_id":123,"template":"welcome"}`),
		RunAt:          time.Now().Add(5 * time.Minute),
		IdempotencyKey: "email:welcome:123",
	})
	if err != nil {
		panic(err)
	}

	scheduler, err := redq.NewScheduler(client, redq.SchedulerOptions{
		Namespace:  "redq",
		Queue:      queue,
		InstanceID: "scheduler-1",
		BatchSize:  100,
		LeaseTTL:   10 * time.Second,
	})
	if err != nil {
		panic(err)
	}

	consumer, err := redq.NewConsumer(client, EmailHandler{}, redq.ConsumerOptions{
		Namespace:       "redq",
		Queue:           queue,
		Group:           "email-workers",
		ConsumerName:    "consumer-1",
		ReadCount:       10,
		Concurrency:     8,
		PendingMinIdle:  30 * time.Second,
		RetryPolicy:     redq.FixedRetryPolicy{Interval: time.Minute},
		PayloadCleanup:  redq.PayloadCleanupOnSuccess,
		StreamRetention: redq.StreamRetentionDeleteOnAck,
		MissingPayload:  redq.MissingPayloadMoveToDLQ,
	})
	if err != nil {
		panic(err)
	}

	go func() { _ = scheduler.Start(ctx) }()
	go func() { _ = consumer.Start(ctx) }()

	fmt.Println("enqueued", result.TaskID)
	select {}
}

Core API

Producer
producer, err := redq.NewProducer(redisClient, redq.ProducerOptions{
	Namespace: "redq",
	Queue:     redq.DefaultQueueConfig("email"),
})

result, err := producer.Enqueue(ctx, redq.Task{
	Payload:        []byte(`{"kind":"welcome"}`),
	RunAt:          time.Now().Add(10 * time.Minute),
	MaxAttempts:    3,
	IdempotencyKey: "email:welcome:123",
})

Cancel a task that has not yet reached the stream:

cancelResult, err := producer.Cancel(ctx, result.TaskID)

Cancel only removes tasks that are still in the delay ZSet. Once a task has entered a Redis Stream, it should be handled with application-level idempotency.

Scheduler
scheduler, err := redq.NewScheduler(redisClient, redq.SchedulerOptions{
	Namespace:  "redq",
	Queue:      redq.DefaultQueueConfig("email"),
	InstanceID: "scheduler-1",
	BatchSize:  100,
	LeaseTTL:   10 * time.Second,
})

err = scheduler.Start(ctx)

Schedulers can run in multiple instances. Shard leases ensure that only one scheduler owns a shard at a time.

Consumer
type Handler struct{}

func (Handler) Handle(ctx context.Context, task *redq.Task) error {
	// Make this operation idempotent.
	return nil
}

consumer, err := redq.NewConsumer(redisClient, Handler{}, redq.ConsumerOptions{
	Namespace:      "redq",
	Queue:          redq.DefaultQueueConfig("email"),
	Group:          "email-workers",
	ConsumerName:   "consumer-1",
	ReadCount:      10,
	Concurrency:    8,
	PendingMinIdle: 30 * time.Second,
	RetryPolicy:    redq.ExponentialRetryPolicy{
		Initial:    time.Second,
		Max:        time.Minute,
		Multiplier: 2,
	},
})

err = consumer.Start(ctx)
DLQ
dlq, err := redq.NewDLQ(redisClient, redq.DLQOptions{
	Namespace: "redq",
	Queue:     redq.DefaultQueueConfig("email"),
})

messages, err := dlq.ListMessages(ctx, redq.DLQListOptions{
	Shard: 0,
	Count: 100,
})

results, err := dlq.ReplayBatch(ctx, redq.DLQReplayBatchOptions{
	Shard: 0,
	Count: 100,
	RunAt: time.Now().Add(time.Minute),
})

Delivery Semantics

redq provides at-least-once delivery. It does not promise exactly-once execution.

Applications must make handlers idempotent. Good idempotency inputs include:

  • Task.ID
  • Task.IdempotencyKey
  • a business identifier inside Task.Payload

Duplicate delivery can happen when:

  • a handler succeeds but the process crashes before XACK;
  • a pending message is reclaimed by another consumer;
  • a network error makes a Redis command result ambiguous;
  • handler execution exceeds the configured pending idle threshold.

Redis Data Model

All keys for a queue shard share the same Redis Cluster hash tag:

redq:{<queue>:<shard>}:delay
redq:{<queue>:<shard>}:payload
redq:{<queue>:<shard>}:stream
redq:{<queue>:<shard>}:dlq
redq:{<queue>:<shard>}:lease
redq:{<queue>:<shard>}:meta

Example:

redq:{email:3}:delay
redq:{email:3}:payload
redq:{email:3}:stream
redq:{email:3}:dlq
redq:{email:3}:lease

The hash tag keeps Lua scripts and multi-key operations inside one Redis Cluster slot.

Operational Notes

  • Run at least one scheduler for every active queue.
  • Use unique ConsumerName values per live consumer instance. Leaving it empty lets redq generate one.
  • Keep PendingMinIdle greater than the normal maximum handler duration.
  • Keep handlers reentrant and idempotent.
  • Do not change a queue's shard count casually after production traffic starts.
  • PayloadCleanupOnSuccess is the default and prevents payload hashes from growing forever.
  • StreamRetentionDeleteOnAck is the default and prevents Redis Streams from growing forever.
  • Use PayloadCleanupNever or StreamRetentionKeep only when you have a separate retention plan.
  • Missing payloads become DLQ tombstones by default. They are queryable but cannot be replayed.

Project Layout

.
├── *.go                  Public redq package API
├── internal/lua          Redis Lua scripts
├── internal/rediskey     Redis key builder and Cluster slot helpers
├── internal/shard        Deterministic shard selection
├── internal/util         Internal utilities
├── cmd/examples          Runnable examples
├── tests/integration     Real Redis integration tests
└── docs                  Architecture notes and verification records

Development

Run all tests:

go test ./...

Run race-sensitive package tests:

go test -race . ./internal/rediskey ./internal/shard -count=1

Run real Redis integration tests:

DELAYQUEUE_REDIS_ADDR=localhost:6379 DELAYQUEUE_REDIS_PASSWORD=123456 go test ./tests/integration -count=1 -v

Run benchmarks:

go test -run '^$' -bench=. ./internal/rediskey ./internal/shard .

Documentation

License

MIT. See LICENSE.

Documentation

Overview

Package redq provides a Redis-backed delay queue for Go services.

It stores delayed tasks in Redis ZSets, schedules due tasks into Redis Streams, consumes them through Consumer Groups, and includes retry, DLQ, pending recovery, and Redis Cluster-safe key semantics.

Index

Constants

View Source 
const (
	DefaultConsumerGroup        = "default"
	DefaultConsumerReadCount    = 10
	DefaultConsumerPollInterval = 200 * time.Millisecond
	DefaultConsumerConcurrency  = 1
	DefaultPendingMinIdle       = 30 * time.Second
)
View Source 
const (
	DefaultNamespace       = "redq"
	DefaultQueueName       = "default"
	DefaultShardCount      = 16
	DefaultMaxPayloadBytes = 256 * 1024
)
View Source 
const (
	DefaultSchedulerBatchSize       = 100
	DefaultSchedulerTickInterval    = 200 * time.Millisecond
	DefaultSchedulerLeaseTTL        = 10 * time.Second
	DefaultSchedulerShutdownTimeout = 5 * time.Second
)
View Source 
const DefaultDLQListCount = 100
View Source 
const DefaultMaxAttempts = 3
View Source 
const DefaultRetryInterval = time.Second

Variables

View Source 
var (
	// ErrStaleMessage 表示当前 Consumer 已不再拥有这条 pending 消息，框架会跳过本次 ack/retry/DLQ 状态写入。
	ErrStaleMessage = errors.New("redq: stale stream message")
	// ErrMissingPayload 表示 Stream 消息存在，但 payload hash 中已找不到对应任务内容。
	ErrMissingPayload = errors.New("redq: task payload missing")
)
View Source 
var ErrDLQMessageNotFound = errors.New("redq: dlq message not found")

Functions

func NewUUID 

func NewUUID() (string, error)

NewUUID 生成 UUIDv7 字符串，作为默认任务 ID。

func ValidateName 

func ValidateName(field string, name string) error

func ValidateQueueName 

func ValidateQueueName(name string) error

Types

type CancelResult 

type CancelResult struct {
	TaskID   string // 被请求取消的任务 ID。
	Queue    string // Producer 配置的队列名。
	Shard    int    // 任务所属 shard。
	Canceled bool   // true 表示任务仍在 delay ZSet 中，并已被删除。
}

CancelResult 表示一次未到期任务取消的结果。

type CancelableProducer 

type CancelableProducer interface {
	Producer
	Canceler
}

type Canceler 

type Canceler interface {
	Cancel(ctx context.Context, taskID string) (*CancelResult, error)
}

type Config 

type Config struct {
	Namespace string
	Redis     RedisConfig
	Queues    []QueueConfig
}

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig 返回框架默认配置。

func (Config) Validate 

func (c Config) Validate() error

Validate 校验全局配置。

type Consumer 

type Consumer interface {
	Start(ctx context.Context) error
}

Consumer 表示消费端运行入口。

type ConsumerClient 

type ConsumerClient interface {
	Eval(ctx context.Context, script string, keys []string, args ...interface{}) *redis.Cmd
	XGroupCreateMkStream(ctx context.Context, stream, group, start string) *redis.StatusCmd
	XReadGroup(ctx context.Context, args *redis.XReadGroupArgs) *redis.XStreamSliceCmd
	XAutoClaim(ctx context.Context, args *redis.XAutoClaimArgs) *redis.XAutoClaimCmd
	HGet(ctx context.Context, key, field string) *redis.StringCmd
}

ConsumerClient 是 Consumer 需要的最小 Redis 能力，便于单元测试替换。

type ConsumerOptions 

type ConsumerOptions struct {
	Namespace       string
	Queue           QueueConfig
	Group           string
	ConsumerName    string
	ReadCount       int
	Concurrency     int
	PollInterval    time.Duration
	PendingMinIdle  time.Duration
	RecoveryCount   int
	RetryPolicy     RetryPolicy
	PayloadCleanup  PayloadCleanupPolicy
	StreamRetention StreamRetentionPolicy
	MissingPayload  MissingPayloadPolicy
	Now             func() time.Time
}

ConsumerOptions 配置 Consumer Group、读取批量、并发、pending recovery、重试、payload cleanup、Stream retention 和 payload 缺失策略。

func DefaultConsumerOptions 

func DefaultConsumerOptions(queue string) ConsumerOptions

DefaultConsumerOptions 返回 Consumer 默认配置。

func (ConsumerOptions) Validate 

func (o ConsumerOptions) Validate() error

type DLQClient 

type DLQClient interface {
	Eval(ctx context.Context, script string, keys []string, args ...interface{}) *redis.Cmd
	XRangeN(ctx context.Context, stream, start, stop string, count int64) *redis.XMessageSliceCmd
	HGet(ctx context.Context, key, field string) *redis.StringCmd
}

DLQClient 是 DLQ 管理 API 需要的最小 Redis 能力，便于单元测试替换。

type DLQListOptions 

type DLQListOptions struct {
	Shard int
	Start string
	Stop  string
	Count int64
}

DLQListOptions 配置 DLQ 查询范围。

func (DLQListOptions) Validate 

func (o DLQListOptions) Validate() error

type DLQMessage 

type DLQMessage struct {
	ID       string
	Task     Task
	Shard    int
	Error    string
	FailedAt time.Time
}

DLQMessage 表示一条可人工查看或重放的死信消息。

type DLQOptions 

type DLQOptions struct {
	Namespace string
	Queue     QueueConfig
	Now       func() time.Time
}

DLQOptions 配置 DLQ 管理器。

func DefaultDLQOptions 

func DefaultDLQOptions(queue string) DLQOptions

DefaultDLQOptions 返回 DLQ 管理器默认配置。

func (DLQOptions) Validate 

func (o DLQOptions) Validate() error

type DLQReplayBatchOptions 

type DLQReplayBatchOptions struct {
	Shard int
	Start string
	Stop  string
	Count int64
	RunAt time.Time
}

DLQReplayBatchOptions 配置 DLQ 批量 replay。

func (DLQReplayBatchOptions) Validate 

func (o DLQReplayBatchOptions) Validate() error

type DLQReplayResult 

type DLQReplayResult struct {
	TaskID   string
	Queue    string
	Shard    int
	RunAt    time.Time
	Replayed bool
}

DLQReplayResult 表示一次 DLQ replay 的结果。

type EnqueueResult 

type EnqueueResult struct {
	TaskID    string // 最终任务 ID。
	Queue     string // 最终队列名。
	Shard     int    // 任务所属 shard。
	Duplicate bool   // true 表示 Redis 中已存在同 task ID 的 payload，未覆盖原任务。
}

EnqueueResult 表示 Producer 入队结果。

type ExponentialRetryPolicy 

type ExponentialRetryPolicy struct {
	Initial    time.Duration
	Max        time.Duration
	Multiplier float64
}

ExponentialRetryPolicy 表示指数退避重试策略。

func (ExponentialRetryPolicy) NextDelay 

func (p ExponentialRetryPolicy) NextDelay(attempt int) time.Duration

type FixedRetryPolicy 

type FixedRetryPolicy struct {
	Interval time.Duration
}

FixedRetryPolicy 表示固定间隔重试策略。

func (FixedRetryPolicy) NextDelay 

func (p FixedRetryPolicy) NextDelay(int) time.Duration

type Handler 

type Handler interface {
	Handle(ctx context.Context, task *Task) error
}

Handler 是业务处理入口；实现方必须保证幂等。

type IDGenerator 

type IDGenerator func() (string, error)

IDGenerator 定义任务 ID 生成函数，业务可以注入自己的 ID 策略。

type MissingPayloadPolicy 

type MissingPayloadPolicy int
const (
	// MissingPayloadDefault 使用框架默认策略，目前等价于 MissingPayloadMoveToDLQ。
	MissingPayloadDefault MissingPayloadPolicy = iota
	// MissingPayloadMoveToDLQ 表示 payload 丢失时 ack 原消息，并写入一条无 payload 的 DLQ tombstone。
	MissingPayloadMoveToDLQ
	// MissingPayloadKeepPending 表示 payload 丢失时返回错误并保留 pending，便于人工排查。
	MissingPayloadKeepPending
)

type PayloadCleanupPolicy 

type PayloadCleanupPolicy int
const (
	// PayloadCleanupDefault 使用框架默认策略，目前等价于 PayloadCleanupOnSuccess。
	PayloadCleanupDefault PayloadCleanupPolicy = iota
	// PayloadCleanupOnSuccess 表示 Handler 成功且 ack 成功后删除 payload。
	PayloadCleanupOnSuccess
	// PayloadCleanupNever 表示成功消费后保留 payload，适合审计场景，但需要业务自行处理 retention。
	PayloadCleanupNever
)

type Producer 

type Producer interface {
	Enqueue(ctx context.Context, task Task) (*EnqueueResult, error)
}

type ProducerClient 

type ProducerClient interface {
	Eval(ctx context.Context, script string, keys []string, args ...interface{}) *redis.Cmd
}

ProducerClient 是 Producer 需要的最小 Redis 能力，便于单元测试用 fake client 替换。

type ProducerOptions 

type ProducerOptions struct {
	Namespace   string
	Queue       QueueConfig
	IDGenerator IDGenerator
	Now         func() time.Time
}

ProducerOptions 配置 Producer 的 namespace、queue、ID 生成器和时间来源。

func DefaultProducerOptions 

func DefaultProducerOptions(queue string) ProducerOptions

DefaultProducerOptions 返回生产者默认配置。

func (ProducerOptions) Validate 

func (o ProducerOptions) Validate() error

type QueueConfig 

type QueueConfig struct {
	Name            string
	Shards          int
	MaxPayloadBytes int
	DefaultAttempts int
}

QueueConfig 描述一个队列的分片数、payload 大小上限和默认尝试次数。

func DefaultQueueConfig 

func DefaultQueueConfig(name string) QueueConfig

DefaultQueueConfig 返回指定队列名的默认配置。

func (QueueConfig) Validate 

func (c QueueConfig) Validate() error

Validate 校验队列配置。

type RedisClient 

type RedisClient = redis.UniversalClient

RedisClient 是 go-redis UniversalClient 的别名。

func NewRedisClient 

func NewRedisClient(config RedisConfig) (RedisClient, error)

NewRedisClient 根据 RedisConfig 创建 go-redis UniversalClient。

type RedisConfig 

type RedisConfig struct {
	Addrs        []string
	Username     string
	Password     string
	DB           int
	Protocol     int
	DialTimeout  time.Duration
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	PoolSize     int
	MinIdleConns int
}

RedisConfig 描述 Redis 连接配置。

func DefaultRedisConfig 

func DefaultRedisConfig() RedisConfig

DefaultRedisConfig 返回本地 Redis 默认配置。

func (RedisConfig) UniversalOptions 

func (c RedisConfig) UniversalOptions() *redis.UniversalOptions

UniversalOptions 转换为 go-redis UniversalOptions。

func (RedisConfig) Validate 

func (c RedisConfig) Validate() error

Validate 校验 Redis 配置。

type RedisConsumer 

type RedisConsumer struct {
	// contains filtered or unexported fields
}

RedisConsumer 是基于 Redis Stream Consumer Group 的消费端实现。

func NewConsumer 

func NewConsumer(client ConsumerClient, handler Handler, options ConsumerOptions) (*RedisConsumer, error)

NewConsumer 创建 RedisConsumer。

func (*RedisConsumer) RecoverPendingOnce 

func (c *RedisConsumer) RecoverPendingOnce(ctx context.Context) (int, error)

RecoverPendingOnce 执行一轮 pending 消息恢复，返回本轮成功恢复并处理的消息数。

func (*RedisConsumer) RunOnce 

func (c *RedisConsumer) RunOnce(ctx context.Context) (int, error)

RunOnce 执行一轮普通 Stream 消费，返回本轮成功处理的消息数。

func (*RedisConsumer) Start 

func (c *RedisConsumer) Start(ctx context.Context) error

Start 持续运行普通消费和 pending recovery，直到 ctx 结束。

type RedisDLQ 

type RedisDLQ struct {
	// contains filtered or unexported fields
}

RedisDLQ 提供 DLQ 查询和 replay 能力。

func NewDLQ 

func NewDLQ(client DLQClient, options DLQOptions) (*RedisDLQ, error)

NewDLQ 创建 RedisDLQ。

func (*RedisDLQ) ListMessages 

func (d *RedisDLQ) ListMessages(ctx context.Context, options DLQListOptions) ([]DLQMessage, error)

ListMessages 查询指定 shard 的 DLQ 消息，并从 payload hash 还原 Task。

func (*RedisDLQ) ReplayBatch 

func (d *RedisDLQ) ReplayBatch(ctx context.Context, options DLQReplayBatchOptions) ([]DLQReplayResult, error)

ReplayBatch 批量重放指定范围内的 DLQ 消息。

func (*RedisDLQ) ReplayMessage 

func (d *RedisDLQ) ReplayMessage(ctx context.Context, shard int, dlqMessageID string, runAt time.Time) (*DLQReplayResult, error)

ReplayMessage 重放单条 DLQ 消息；默认会重置 Attempt 和 LastError。

type RedisProducer 

type RedisProducer struct {
	// contains filtered or unexported fields
}

RedisProducer 是基于 Redis ZSet + payload hash 的 Producer 实现。

func NewProducer 

func NewProducer(client ProducerClient, options ProducerOptions) (*RedisProducer, error)

NewProducer 创建 RedisProducer。

func (*RedisProducer) Cancel 

func (p *RedisProducer) Cancel(ctx context.Context, taskID string) (*CancelResult, error)

Cancel 删除尚未到期的延迟任务。已经进入 Stream 或不存在的任务会返回 Canceled=false。

func (*RedisProducer) Enqueue 

func (p *RedisProducer) Enqueue(ctx context.Context, task Task) (*EnqueueResult, error)

Enqueue 校验任务、选择 shard，并通过 Lua 原子写入 payload hash 和 delay ZSet。

type RedisScheduler 

type RedisScheduler struct {
	// contains filtered or unexported fields
}

RedisScheduler 负责抢占 shard lease，并把到期任务从 delay ZSet 移动到 Stream。

func NewScheduler 

func NewScheduler(client SchedulerClient, options SchedulerOptions) (*RedisScheduler, error)

NewScheduler 创建 RedisScheduler。

func (*RedisScheduler) RunOnce 

func (s *RedisScheduler) RunOnce(ctx context.Context) (int, error)

RunOnce 执行一轮所有 shard 的 lease claim 和到期任务调度，主要用于测试和受控调度。

func (*RedisScheduler) Start 

func (s *RedisScheduler) Start(ctx context.Context) error

Start 持续运行调度循环，直到 ctx 结束。

type RetryPolicy 

type RetryPolicy interface {
	NextDelay(attempt int) time.Duration
}

RetryPolicy 计算第 attempt 次失败后的下一次重试延迟。

type Scheduler 

type Scheduler interface {
	Start(ctx context.Context) error
}

type SchedulerClient 

type SchedulerClient interface {
	Eval(ctx context.Context, script string, keys []string, args ...interface{}) *redis.Cmd
}

SchedulerClient 是 Scheduler 需要的最小 Redis 能力，便于单元测试替换。

type SchedulerOptions 

type SchedulerOptions struct {
	Namespace       string
	Queue           QueueConfig
	InstanceID      string
	BatchSize       int
	TickInterval    time.Duration
	LeaseTTL        time.Duration
	ShutdownTimeout time.Duration
	Now             func() time.Time
}

SchedulerOptions 配置 Scheduler 的 shard lease、批量大小和调度周期。

func DefaultSchedulerOptions 

func DefaultSchedulerOptions(queue string) SchedulerOptions

DefaultSchedulerOptions 返回 Scheduler 默认配置。

func (SchedulerOptions) Validate 

func (o SchedulerOptions) Validate() error

type StreamRetentionPolicy 

type StreamRetentionPolicy int
const (
	// StreamRetentionDefault 使用框架默认策略，目前等价于 StreamRetentionDeleteOnAck。
	StreamRetentionDefault StreamRetentionPolicy = iota
	// StreamRetentionDeleteOnAck 表示消息 ack 成功后删除原 Stream entry，避免 Stream 长期无限增长。
	StreamRetentionDeleteOnAck
	// StreamRetentionKeep 表示 ack 后保留原 Stream entry，适合需要自行审计或外部裁剪的场景。
	StreamRetentionKeep
)

type Task 

type Task struct {
	ID             string            // 任务唯一 ID；为空时 Producer 使用默认 IDGenerator 生成 UUIDv7。
	Queue          string            // 队列名；为空时使用 Producer/Consumer 配置中的默认队列。
	Payload        []byte            // 业务 payload；框架只负责存储和透传。
	RunAt          time.Time         // 期望投递时间；为空时按当前时间立即调度。
	MaxAttempts    int               // 最大尝试次数；小于等于 0 时使用默认值。
	Attempt        int               // 当前已尝试次数；由框架在 retry/DLQ 路径维护。
	IdempotencyKey string            // 业务幂等 key；仅作为元数据透传，不参与框架内部分片选择。
	Metadata       map[string]string // 透传元数据，可放 trace、tenant 等轻量字段。
	LastError      string            // 最近一次 Handler 错误；由框架在失败路径维护。
}

Task 表示一个延迟任务，是 Producer 写入和 Handler 消费的基本单位。 业务侧应使用 ID、IdempotencyKey 或 payload 中的业务主键实现幂等。

func (Task) EffectiveMaxAttempts 

func (t Task) EffectiveMaxAttempts() int

EffectiveMaxAttempts 返回任务实际使用的最大尝试次数。

func (Task) Validate 

func (t Task) Validate() error

Validate 校验任务是否具备入队所需的最小字段。

type ValidationError 

type ValidationError struct {
	Field  string
	Reason string
}

ValidationError 表示配置或任务参数校验失败。

func (ValidationError) Error 

func (e ValidationError) Error() string

Source Files

View all Source files

Directories

Path Synopsis
cmd
examples/basic command
internal
lua
rediskey
shard
util

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.