Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

queue

package

Go to main page

Versions in this module

v1

v1.4.0

Jun 18, 2026

v1.3.0

Jun 18, 2026

v1.2.0

Jun 18, 2026

v1.1.0

Jun 18, 2026

v1.0.0

Jun 10, 2026

Changes in this version

+ var ErrEmptyQueue = errors.New("queue name cannot be empty")

+ var ErrNilTask = errors.New("task cannot be nil")

+ var ErrQueueClosed = errors.New("queue is closed")

+ var ErrTaskNotFound = errors.New("task not found")

+ type Handler interface

+ Handle func(ctx context.Context, task *Task) error

+ type HandlerFunc func(ctx context.Context, task *Task) error

+ func (f HandlerFunc) Handle(ctx context.Context, task *Task) error

+ type Queue interface

+ Ack func(ctx context.Context, taskID string) error

+ Close func() error

+ Dequeue func(ctx context.Context) (*Task, error)

+ Enqueue func(ctx context.Context, task *Task) error

+ EnqueueWithDelay func(ctx context.Context, task *Task, delay time.Duration) error

+ GetStats func(ctx context.Context) (*QueueStats, error)

+ GetTask func(ctx context.Context, taskID string) (*Task, error)

+ Nack func(ctx context.Context, taskID string, taskErr error) error

+ Purge func(ctx context.Context) error

+ type QueueConfig struct

+ DeadLetterMaxRetries int

+ DelayQueueScanInterval time.Duration

+ EnableDeadLetter bool

+ MaxConcurrency int

+ MaxRetries int

+ Name string

+ Prefix string

+ ProcessingCleanupInterval time.Duration

+ RetryDelay RetryDelayStrategy

+ VisibilityTimeout int

+ func DefaultConfig() QueueConfig

+ func (c QueueConfig) WithDeadLetter(enable bool, maxRetries int) QueueConfig

+ func (c QueueConfig) WithMaxConcurrency(max int) QueueConfig

+ func (c QueueConfig) WithMaxRetries(retries int) QueueConfig

+ func (c QueueConfig) WithName(name string) QueueConfig

+ func (c QueueConfig) WithRetryDelay(strategy RetryDelayStrategy) QueueConfig

+ func (c QueueConfig) WithScanIntervals(delayScan, cleanup time.Duration) QueueConfig

+ func (c QueueConfig) WithVisibilityTimeout(timeout int) QueueConfig

+ type QueueManager struct

+ func NewQueueManager(client *redis.Client) *QueueManager

+ func (m *QueueManager) EnqueueTask(ctx context.Context, queueName string, task *Task) error

+ func (m *QueueManager) EnqueueTaskWithDelay(ctx context.Context, queueName string, task *Task, delay time.Duration) error

+ func (m *QueueManager) GetOrCreateQueue(name string, config QueueConfig) *RedisQueue

+ func (m *QueueManager) GetQueueStats(ctx context.Context) (map[string]*QueueStats, error)

+ func (m *QueueManager) RegisterWorkerPool(queueName string, handler Handler, config QueueConfig) (*WorkerPoolImpl, error)

+ func (m *QueueManager) StartAllWorkerPools(ctx context.Context) error

+ func (m *QueueManager) StopAllWorkerPools() error

+ type QueueStats struct

+ Completed int64

+ DeadLetter int64

+ Delayed int64

+ Failed int64

+ Name string

+ Pending int64

+ Processing int64

+ type RedisQueue struct

+ func NewRedisQueue(client *redis.Client, config QueueConfig) *RedisQueue

+ func (q *RedisQueue) Ack(ctx context.Context, taskID string) error

+ func (q *RedisQueue) Close() error

+ func (q *RedisQueue) Dequeue(ctx context.Context) (*Task, error)

+ func (q *RedisQueue) Enqueue(ctx context.Context, task *Task) error

+ func (q *RedisQueue) EnqueueWithDelay(ctx context.Context, task *Task, delay time.Duration) error

+ func (q *RedisQueue) GetStats(ctx context.Context) (*QueueStats, error)

+ func (q *RedisQueue) GetTask(ctx context.Context, taskID string) (*Task, error)

+ func (q *RedisQueue) Nack(ctx context.Context, taskID string, taskErr error) error

+ func (q *RedisQueue) Purge(ctx context.Context) error

+ type RetryDelayStrategy string

+ const RetryDelayExponential

+ const RetryDelayFixed

+ const RetryDelayRandom

+ type Task struct

+ CompletedAt int64

+ CreatedAt int64

+ DelayUntil int64

+ Error string

+ ID string

+ MaxRetries int

+ Metadata map[string]string

+ Payload []byte

+ Queue string

+ RetryCount int

+ StartedAt int64

+ Status TaskStatus

+ Type string

+ func NewTask(queue, taskType string, payload []byte) *Task

+ func UnmarshalTask(data []byte) (*Task, error)

+ func (t *Task) Marshal() ([]byte, error)

+ func (t *Task) WithDelay(delay time.Duration) *Task

+ func (t *Task) WithMaxRetries(maxRetries int) *Task

+ func (t *Task) WithMetadata(key, value string) *Task

+ type TaskStatus string

+ const TaskStatusCompleted

+ const TaskStatusDeadLetter

+ const TaskStatusFailed

+ const TaskStatusPending

+ const TaskStatusProcessing

+ type Worker struct

+ func NewWorker(id string, queue Queue, handler Handler) *Worker

+ func (w *Worker) Start(ctx context.Context) error

+ func (w *Worker) Stop() error

+ func (w *Worker) WithLogger(log logger.Logger) *Worker

+ type WorkerPool interface

+ IsRunning func() bool

+ Start func(ctx context.Context) error

+ Stop func() error

+ type WorkerPoolImpl struct

+ func NewWorkerPool(queue Queue, handler Handler, config QueueConfig) *WorkerPoolImpl

+ func (p *WorkerPoolImpl) IsRunning() bool

+ func (p *WorkerPoolImpl) Start(ctx context.Context) error

+ func (p *WorkerPoolImpl) Stop() error

+ func (p *WorkerPoolImpl) WithLogger(log logger.Logger) *WorkerPoolImpl

v0

v0.0.3

Jun 9, 2026

v0.0.2

Jun 4, 2026

v0.0.1

Jun 4, 2026