Getting Started

This page walks you from zero to a running, horizontally-scalable task queue. You will register a task, start a worker, dispatch messages, and observe them being processed.

Requirements

Requirement	Detail
Go	1.24.9 or newer (see go.mod). The module uses recent generics and stdlib features.
Redis	6.0 minimum. 6.2+ recommended -- XPENDING's IDLE filter (used for efficient pending reclaims) needs >= 6.2; older versions fall back to client-side filtering.
Network	Each TaskRunner instance needs TCP reachability to Redis. No peer-to-peer networking between instances -- they coordinate entirely through Redis.

No Redis for local dev?

The test suite (and the examples' tests) use miniredis, an in-process Redis compatible server. You do not need a real Redis to run go test ./....

Installation

go get github.com/soroosh-tanzadeh/taskrunner

TaskRunner ships its dependencies vendored in vendor/, so if your module is in vendor mode (GOFLAGS=-mod=vendor) it will just work. Otherwise go mod will resolve them normally.

The five-minute quickstart

This is the canonical "process a queue" program. It registers one task, starts the runner, dispatches 100 messages, and then shuts down cleanly.

package main

import (
    "context"
    "fmt"
    "strconv"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/soroosh-tanzadeh/taskrunner/redisstream"
    "github.com/soroosh-tanzadeh/taskrunner/runner"
)

func main() {
    rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379", DB: 0, PoolSize: 50})

    // 1. Build the queue backend (Redis Streams, the shipped implementation).
    queue := redisstream.NewRedisStreamMessageQueueWithOptions(
        rdb,
        redisstream.WithPrefix("example"),
        redisstream.WithQueue("tasks"),
        redisstream.WithReClaimDelay(30*time.Second),
        redisstream.WithDeleteOnAck(true),
    )

    // 2. Build the runner.
    tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
        BatchSize:          10,
        ConsumerGroup:      "example",
        ConsumersPrefix:    "default",
        NumWorkers:         8,
        NumFetchers:        4,
        LongQueueHook:      func(s runner.Stats) { fmt.Printf("%+v\n", s) },
        LongQueueThreshold: 30 * time.Second,
    }, rdb, queue)

    // 3. Register the task and its handler.
    tr.RegisterTask(&runner.Task{
        Name:     "exampletask",
        MaxRetry: 5,
        Action: func(ctx context.Context, payload any) error {
            fmt.Printf("Hello from example task %v\n", payload)
            return nil
        },
    })

    // 4. Start the runner in the background. Start() blocks until ctx is canceled.
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() { defer wg.Done(); _ = tr.Start(ctx) }()

    // 5. Dispatch work.
    for i := 0; i < 100; i++ {
        _ = tr.Dispatch(context.Background(), "exampletask", strconv.Itoa(i))
    }

    time.Sleep(2 * time.Second)
    cancel() // graceful shutdown
    wg.Wait()
}

What just happened?

1. redisstream.NewRedisStreamMessageQueueWithOptions wires a Redis Streams queue. WithDeleteOnAck(true) deletes the stream entry after processing so the stream never grows unbounded. WithReClaimDelay(30s) is the visibility timeout -- a message that has not been acked within 30s becomes eligible for reclaim by another consumer.
2. NewTaskRunner constructs the engine. It does not connect to Redis eagerly; the first network call happens when Start() runs.
3. RegisterTask adds the task to an in-memory registry keyed by name. The registry is a safemap.SafeMap, read on every Dispatch and every worker invocation.
4. Start(ctx) spins up NumFetchers fetcher goroutines, an ants worker pool of NumWorkers, the replication registrar, the timing-aggregation loop, and leader election. It blocks until ctx is canceled.
5. Dispatch looks up the task, builds a TaskMessage, JSON-marshals it, and XADDs it to the stream. Workers pick it up via XREADGROUP.

Running multiple instances

Horizontal scaling is automatic -- just run the same binary more than once, pointed at the same Redis and the same ConsumerGroup. Redis Streams consumer-group semantics ensure each message is delivered to exactly one consumer in the group.

# Terminal 1
REDIS_ADDR=127.0.0.1:6379 ./myapp

# Terminal 2 -- shares the load
REDIS_ADDR=127.0.0.1:6379 ./myapp

One of the instances will win leader election. Only the leader runs the delayed scheduler, computes timing stats, and publishes worker-tuning capacity. Followers apply the leader's published capacity. The workload (message processing) is shared across all instances.

Next steps

• Architecture -- the fetcher/worker split, the message lifecycle, and how heartbeats prevent double-processing.
• Delayed Scheduling -- run jobs in the future with DispatchDelayed and ScheduleFor.
• Unique Jobs -- prevent duplicate enqueues.
• Worker Tuning -- let the leader autoscale the worker pool based on predicted wait time.
• Configuration Reference -- every field, defaults, and tuning advice.