groupmq

Go-GroupMQ

English | 简体中文

A fast, reliable Redis-backed per-group FIFO queue for Go with guaranteed job ordering and parallel processing across groups.

Ported from GroupMQ (TypeScript/Node.js) - a BullMQ-inspired queue library optimized for per-group sequential processing.

Features

• Per-group FIFO ordering - Jobs within the same group process in strict order, perfect for user workflows, data pipelines, and sequential operations
• Parallel processing across groups - Process multiple groups simultaneously while maintaining order within each group
• High performance - Optimized with goroutines and atomic Lua scripts for Redis
• Built-in ordering strategies - Handle out-of-order job arrivals with OrderingNone, OrderingScheduler, or OrderingInMemory methods
• Automatic recovery - Stalled job detection and connection error handling
• Production ready - Atomic operations, graceful shutdown, and comprehensive logging
• Zero polling - Efficient blocking operations prevent wasteful Redis calls
• CLI monitor tool - Real-time queue monitoring and debugging

Installation

go get github.com/bootcs-dev/go-groupmq

Quick Start

package main

import (
	"context"
	"fmt"
	"log"

	groupmq "github.com/bootcs-dev/go-groupmq"
	"github.com/redis/go-redis/v9"
)

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

	// Connect to Redis
	rdb := redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})
	defer rdb.Close()

	// Create queue
	queue := groupmq.NewQueue(rdb, groupmq.QueueOptions{
		Namespace:          "orders",      // Will be prefixed with 'groupmq:'
		JobTimeoutMs:       30000,         // How long before job times out
		DefaultMaxAttempts: 3,             // Default max retry attempts
		Logger:             groupmq.NewDefaultLogger(true, "orders"),
	})

	// Add a job
	jobID, err := queue.Add(ctx, groupmq.AddOptions{
		GroupId: "user:42",
		Data: map[string]interface{}{
			"type":   "charge",
			"amount": 999,
		},
		OrderMs:     0,  // Use current timestamp, or provide event timestamp
		MaxAttempts: 5,
	})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Added job: %s\n", jobID)

	// Create worker
	worker := groupmq.NewWorker(queue, func(ctx context.Context, job *groupmq.ReservedJob) (interface{}, error) {
		fmt.Printf("Processing: %+v\n", job.Data)
		// Do work here...
		return "done", nil
	})

	// Run worker (blocks)
	worker.Run(ctx)
}

CLI Monitor Tool

Go-GroupMQ includes a powerful CLI tool for monitoring and debugging your queues in real-time:

Installation

# Install globally
go install github.com/bootcs-dev/go-groupmq/cmd/monitor@latest

# Or run directly
go run github.com/bootcs-dev/go-groupmq/cmd/monitor@latest -n myqueue

Usage

Monitor mode - Real-time queue statistics:

monitor -n orders
monitor -n orders -r redis://localhost:6379 -i 2000

List completed jobs:

monitor -n orders --list-completed --limit 10

List failed jobs:

monitor -n orders --list-failed --limit 20

CLI Options:

• -n, --namespace - Queue namespace (required)
• -r, --redis-url - Redis URL (default: redis://127.0.0.1:6379)
• -i, --interval - Update interval in ms (default: 1000)
• --list-completed - List recent completed jobs
• --list-failed - List recent failed jobs
• --limit - Number of jobs to list (default: 20)
• -h, --help - Show help

Queue Options

type QueueOptions struct {
	Namespace                  string              // Unique queue name (required)
	Logger                     LoggerInterface     // Logger instance (optional)
	JobTimeoutMs               int                 // Job processing timeout (default: 30000ms)
	DefaultMaxAttempts         int                 // Default max retry attempts (default: 3)
	ReserveScanLimit           int                 // Groups to scan when reserving (default: 20)
	KeepCompleted              int                 // Number of completed jobs to retain (default: 0)
	KeepFailed                 int                 // Number of failed jobs to retain (default: 0)
	SchedulerLockTtlMs         int                 // Scheduler lock TTL (default: 1500ms)
	OrderingMethod             OrderingMethod      // Ordering strategy (default: OrderingNone)
	OrderingWindowMs           int                 // Time window for ordering (required for non-none methods)
	OrderingMaxWaitMultiplier  int                 // Max grace period multiplier for in-memory (default: 3)
	OrderingGracePeriodDecay   float64             // Grace period decay factor for in-memory (default: 1.0)
	OrderingMaxBatchSize       int                 // Max jobs to collect in batch for in-memory (default: 10)
}

type OrderingMethod string

const (
	OrderingNone      OrderingMethod = "none"       // No ordering guarantees (fastest)
	OrderingScheduler OrderingMethod = "scheduler"  // Redis buffering for large windows (≥1000ms)
	OrderingInMemory  OrderingMethod = "in-memory"  // Worker collection for small windows (50-500ms)
)

Worker Configuration

// Create worker with handler function
worker := groupmq.NewWorker(queue, handlerFunc)

// Configure worker options
worker.Concurrency = 8           // Process up to 8 jobs in parallel (default: 1)
worker.HeartbeatMs = 5000        // Heartbeat interval (default: max(1000, jobTimeoutMs/3))
worker.MaxStalledCount = 1       // Fail after N stalls (default: 1)
worker.StalledGracePeriod = 0    // Grace period before considering stalled (default: 0)
worker.CleanupIntervalMs = 60000 // Cleanup frequency (default: 60000ms)

// Run worker
go worker.Run(ctx)

// Stop worker gracefully
worker.Stop()

Handler function signature:

type HandlerFunc func(ctx context.Context, job *ReservedJob) (interface{}, error)

Adding Jobs

Basic job

jobID, err := queue.Add(ctx, groupmq.AddOptions{
	GroupId: "user:123",
	Data: map[string]interface{}{
		"action": "send-email",
		"email":  "[email protected]",
	},
})

Job with custom options

jobID, err := queue.Add(ctx, groupmq.AddOptions{
	GroupId:     "user:123",
	Data:        myData,
	OrderMs:     event.CreatedAtMs,  // Timestamp for ordering
	MaxAttempts: 5,                  // Override default max attempts
	JobId:       "custom-id",        // Custom job ID (optional)
	Delay:       3600000,            // Delay in ms (1 hour)
})

Delayed job (run at specific time)

runAt := time.Now().Add(24 * time.Hour)

jobID, err := queue.Add(ctx, groupmq.AddOptions{
	GroupId: "user:123",
	Data:    myData,
	RunAt:   runAt.UnixMilli(),
})

Repeating jobs (cron/interval)

Every N milliseconds:

jobID, err := queue.Add(ctx, groupmq.AddOptions{
	GroupId: "reports",
	Data:    map[string]interface{}{"type": "daily-summary"},
	Repeat: &groupmq.RepeatOptions{
		Every: 5000, // Run every 5 seconds
	},
})

Cron pattern:

jobID, err := queue.Add(ctx, groupmq.AddOptions{
	GroupId: "emails",
	Data:    map[string]interface{}{"type": "weekly-digest"},
	Repeat: &groupmq.RepeatOptions{
		Pattern: "0 9 * * 1-5", // 09:00 Mon-Fri
	},
})

Remove repeating job:

err := queue.RemoveRepeatingJob(ctx, "reports", &groupmq.RepeatOptions{Every: 5000})

Worker Concurrency

Workers support configurable concurrency to process multiple jobs in parallel from different groups:

worker := groupmq.NewWorker(queue, handlerFunc)
worker.Concurrency = 8 // Process up to 8 jobs simultaneously

// Jobs from different groups run in parallel
// Jobs from the same group still run sequentially
go worker.Run(ctx)

Benefits:

• Higher throughput for multi-group workloads
• Efficient resource utilization with goroutines
• Still maintains per-group FIFO ordering

Considerations:

• Each job runs in its own goroutine
• Set concurrency based on job duration and system resources
• Monitor Redis connection pool (go-redis default: 10 connections per client)

Logging

Both Queue and Worker support optional logging:

// Enable default logger
queue := groupmq.NewQueue(rdb, groupmq.QueueOptions{
	Namespace: "orders",
	Logger:    groupmq.NewDefaultLogger(true, "orders"),
})

Custom logger:

type CustomLogger struct{}

func (l *CustomLogger) Debug(msg string, args ...interface{}) { /* ... */ }
func (l *CustomLogger) Info(msg string, args ...interface{})  { /* ... */ }
func (l *CustomLogger) Warn(msg string, args ...interface{})  { /* ... */ }
func (l *CustomLogger) Error(msg string, args ...interface{}) { /* ... */ }

queue := groupmq.NewQueue(rdb, groupmq.QueueOptions{
	Namespace: "orders",
	Logger:    &CustomLogger{},
})

Queue Methods

Job counts and status

// Get all counts at once
counts, err := queue.GetJobCounts(ctx)
// counts: map[string]int64{
//   "active": 5, "waiting": 12, "delayed": 3,
//   "completed": 100, "failed": 2, "total": 122
// }

// Individual counts
activeCount, err := queue.GetActiveCount(ctx)
waitingCount, err := queue.GetWaitingCount(ctx)
delayedCount, err := queue.GetDelayedCount(ctx)
completedCount, err := queue.GetCompletedCount(ctx)
failedCount, err := queue.GetFailedCount(ctx)

Get jobs by status

// Get job IDs
activeJobIDs, err := queue.GetActiveJobs(ctx)
waitingJobIDs, err := queue.GetWaitingJobs(ctx)
delayedJobIDs, err := queue.GetDelayedJobs(ctx)

// Get Job instances
completedJobs, err := queue.GetCompletedJobs(ctx, 20) // returns []*Job
failedJobs, err := queue.GetFailedJobs(ctx, 20)

Group information

// Get all unique groups
groups, err := queue.GetUniqueGroups(ctx) // []string{"user:123", "order:456"}
groupCount, err := queue.GetUniqueGroupsCount(ctx)

// Get jobs in a specific group
jobCount, err := queue.GetGroupJobCount(ctx, "user:123")

Job manipulation

// Get specific job
job, err := queue.GetJob(ctx, jobID)

// Remove job
err = queue.Remove(ctx, jobID)

// Retry failed job
err = queue.Retry(ctx, jobID)

// Promote delayed job to waiting
err = queue.Promote(ctx, jobID)

// Change delay
err = queue.ChangeDelay(ctx, jobID, newDelayMs)

// Update job data
err = queue.UpdateData(ctx, jobID, newData)

Scheduler and maintenance

// Manual scheduler run
err = queue.RunSchedulerOnce(ctx)

// Promote delayed jobs
count, err := queue.PromoteDelayedJobs(ctx)

// Recover stuck groups
count, err := queue.RecoverDelayedGroups(ctx)

// Wait for queue to be empty
isEmpty, err := queue.WaitForEmpty(ctx, timeoutMs)

// Cleanup
err = queue.Close()

Job Instance Methods

job, err := queue.GetJob(ctx, jobID)

// Manipulate the job
err = job.Remove(ctx)
err = job.Retry(ctx)
err = job.Promote(ctx)
err = job.ChangeDelay(ctx, newDelayMs)
err = job.UpdateData(ctx, newData)

// Get job state
state, err := job.GetState(ctx) // "active" | "waiting" | "delayed" | "completed" | "failed"

// Serialize job
jsonData, err := json.Marshal(job)

Graceful Shutdown

// Stop worker gracefully - waits for current jobs to finish
worker.Stop()

// Wait for queue to be empty
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
isEmpty, err := queue.WaitForEmpty(ctx, 30000)

// Recover groups that might be stuck
recoveredCount, err := queue.RecoverDelayedGroups(ctx)

Architecture

Redis Data Structures

Go-GroupMQ uses these Redis keys (all prefixed with groupmq:{namespace}:):

• :g:{groupId} - Sorted set of job IDs in a group, ordered by score
• :ready - Sorted set of group IDs with available jobs
• :job:{jobId} - Hash containing job data
• :lock:{groupId} - String with job ID that owns the group lock (with TTL)
• :processing - Sorted set of active job IDs
• :processing:{jobId} - Hash with processing metadata
• :delayed - Sorted set of delayed jobs
• :completed - Sorted set of completed job IDs
• :failed - Sorted set of failed job IDs
• :repeats - Hash of repeating job definitions

Job Lifecycle

1. Waiting - Job is in :g:{groupId} and group is in :ready
2. Delayed - Job is in :delayed (scheduled for future)
3. Active - Job is in :processing and group is locked
4. Completed - Job is in :completed (retention)
5. Failed - Job exceeded maxAttempts, moved to :failed

Atomic Operations

All critical operations use Lua scripts for atomicity:

• enqueue.lua - Adds job to group queue and ready set
• reserve.lua - Finds ready group, pops head job, locks group
• reserve-batch.lua - Reserves one job from multiple groups
• complete.lua - Marks job complete, unlocks group
• complete-and-reserve-next-with-metadata.lua - Atomic completion + reservation
• retry.lua - Increments attempts, re-enqueues with backoff
• remove.lua - Removes job from all data structures

Ordering and Scoring

Jobs are ordered using a composite score:

score = (orderMs - baseEpoch) * 1000 + seq

• orderMs - User-provided timestamp for event ordering
• baseEpoch - Fixed epoch (1704067200000) to keep scores manageable
• seq - Auto-incrementing sequence for tiebreaking

This ensures:

• Jobs with earlier orderMs process first
• Jobs with same orderMs process in submission order
• Stable, sortable scores
• Daily sequence reset prevents overflow

Performance

Go-GroupMQ is designed for high performance:

Optimizations:

• Goroutines - Lightweight concurrency for parallel job processing
• Batch Operations - ReserveBatch reduces Redis round-trips
• Blocking Operations - Efficient Redis blocking prevents polling
• Lua Scripts - All critical paths are atomic
• Atomic Completion - Complete + reserve next in single operation
• Zero-copy - Minimal data marshaling with Redis hashes

Benchmarks (MacBook M2, Redis 7):

See benchmark/README.md for detailed benchmark results comparing Go-GroupMQ with TypeScript GroupMQ.

Inspiration and Attribution

Go-GroupMQ is a Go port of GroupMQ, which is inspired by BullMQ. We've adapted the TypeScript/Node.js implementation to Go while maintaining the core concepts:

• Per-group FIFO ordering
• Parallel processing across groups
• Atomic Lua script operations
• Flexible ordering strategies

Key differences from TypeScript GroupMQ:

• Native Go concurrency with goroutines
• Strongly typed API with Go structs
• CLI tool compiled to standalone binary
• Similar performance with lower memory footprint

Testing

Requires a local Redis at 127.0.0.1:6379.

# Run all tests
go test -v ./...

# Run with coverage
go test -coverprofile=coverage.out ./...
go tool cover -html=coverage.out

# Run benchmarks
cd benchmark
go run main.go -type=cpu -workers=8 -jobs=10000

Optional Docker Redis:

docker run --rm -p 6379:6379 redis:7

Contributing

Contributions are welcome! Please:

1. Run tests before and after changes
2. Add tests for new features
3. Update documentation
4. Run benchmarks for performance-sensitive changes

License

MIT License - see LICENSE file for details.

Links

• TypeScript version: https://github.com/openpanel-dev/groupmq
• BullMQ (inspiration): https://github.com/taskforcesh/bullmq