VarMQ Worker Configuration Options: Complete Guide to Pool Tuning and Queue Strategy

VarMQ workers expose six core configuration options—concurrency, queue-selection strategy, idle-worker ratios, expiry duration, job ID generation, and lifecycle context—that can be set globally via Default* helpers or per-worker through With* functional options passed to NewWorker.

VarMQ is a high-performance job queue library for Go maintained at goptics/varmq. All worker configuration is centralized in the internal configs struct defined in [config.go](https://github.com/goptics/varmq/blob/main/config.go), enabling precise control over goroutine pools, queue selection algorithms, and automatic resource cleanup.

Core Configuration Options

Concurrency Control

The concurrency option determines how many goroutine workers can process jobs simultaneously. It is stored as uint32 internally but exposed as int in the API.

• Default: 1 (or the number of CPU cores if set to less than 1)
• Global setter: DefaultConcurrency(int)
• Per-worker setter: WithConcurrency(int)
• Runtime adjustment: TunePool(int) in [internal/pool/pool.go](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go)

This value directly controls the size of the internal worker pool managed by the pool subsystem.

Queue Selection Strategy

The strategy option defines the algorithm used when a worker is bound to multiple queues and must select the next job.

• Type: Strategy enum
• Default: Priority
• Available strategies: Priority, RoundRobin, MaxLen, MinLen
• Implementation: Applied by queueManager.next() in [queue_manager.go](https://github.com/goptics/varmq/blob/main/queue_manager.go)

Set globally with DefaultStrategy(Strategy) or per-worker with WithStrategy(Strategy).

Idle Worker Management

Two options control how the pool maintains idle workers for burst capacity:

minIdleWorkerRatio (uint8 percentage):

• Ensures a minimum percentage of the total concurrency remains idle and ready
• Default guarantees at least 1% (minimum one idle worker)
• Calculated via numMinIdleWorkers() as concurrency × minIdleWorkerRatio
• Set globally with DefaultMinIdleWorkerRatio(uint8) or per-worker with WithMinIdleWorkerRatio(uint8)

idleWorkerExpiryDuration (time.Duration):

• Controls automatic pool shrinking when set to a value greater than zero
• Default is unset, meaning exactly one idle worker is always retained indefinitely
• When configured, the background goroutine goRemoveIdleWorkers in [worker.go](https://github.com/goptics/varmq/blob/main/worker.go) periodically removes workers idle longer than this duration
• Set globally with DefaultIdleWorkerExpiryDuration(time.Duration) or per-worker with WithIdleWorkerExpiryDuration(time.Duration)

Job ID Generation

The jobIdGenerator option accepts a function that returns a unique string identifier for each job.

• Type: func() string
• Default: No-op function returning empty string
• Application: Every Job created via loadJobConfigs receives its ID from this generator (stored in jobConfigs.Id)
• Override: Individual jobs can override the generated ID using WithJobId
• Setters: DefaultJobIdGenerator(func() string) globally or WithJobIdGenerator(func() string) per-worker

Lifecycle Context

The ctx option provides a context.Context that drives graceful worker shutdown.

• Default: nil
• Behavior: The internal function goListenToContext in worker.go monitors ctx.Done() and automatically triggers worker.Stop() upon cancellation
• Propagation: Cancellation propagates to all internal goroutines
• Setters: DefaultCtx(context.Context) globally or WithContext(context.Context) per-worker

How to Configure Workers

Setting Global Defaults

Apply configuration values to all workers unless explicitly overridden. The merge logic in loadConfigs / mergeConfigs processes options in order, with later values taking precedence.

varmq.DefaultConcurrency(8)                     // 8 goroutines (or CPU count if <1)
varmq.DefaultStrategy(varmq.RoundRobin)         // round-robin queue selection
varmq.DefaultMinIdleWorkerRatio(20)             // maintain 20% idle workers
varmq.DefaultIdleWorkerExpiryDuration(30 * time.Second)
varmq.DefaultJobIdGenerator(func() string {
    return uuid.NewString()                     // UUID-based job IDs
})
varmq.DefaultCtx(context.Background())

Per-Worker Configuration

Override global defaults for specific workers using functional options passed to NewWorker:

worker := varmq.NewWorker[string](
    func(j varmq.Job[string]) {
        fmt.Println("processing:", j.Payload())
    },

    varmq.WithConcurrency(4),                     // limit to 4 workers
    varmq.WithStrategy(varmq.Priority),          // priority-based selection
    varmq.WithMinIdleWorkerRatio(10),            // 10% minimum idle
    varmq.WithIdleWorkerExpiryDuration(15 * time.Second),
    varmq.WithJobIdGenerator(func() string { 
        return "job-" + strconv.Itoa(rand.Int()) 
    }),
)

if err := worker.Start(); err != nil {
    log.Fatalf("failed to start worker: %v", err)
}

Runtime Pool Tuning

Change the concurrency level dynamically while the worker is running using TunePool:

// Increase pool size to 12 workers during high load
if err := worker.TunePool(12); err != nil {
    log.Printf("cannot tune pool: %v", err)
}

This method updates the internal pool size in internal/pool/pool.go without restarting the worker.

Context-Driven Shutdown

Bind a custom context to enable external control over worker lifecycle:

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

worker := varmq.NewWorker[int](
    func(j varmq.Job[int]) { fmt.Println(j.Payload()) },
    varmq.WithContext(ctx),
)

// Later, signal shutdown across the worker group
cancel() // triggers automatic Stop() via goListenToContext

Summary

• Six configuration options control concurrency, queue strategy, idle ratios, expiry, job IDs, and context lifecycle.
• Two configuration scopes exist: global defaults via Default* functions and per-worker overrides via With* functional options.
• Configuration merging occurs in loadConfigs / mergeConfigs in config.go, applying options sequentially.
• Runtime tuning is supported via TunePool for adjusting concurrency without restarts.
• Automatic cleanup occurs when idleWorkerExpiryDuration is set, driven by goRemoveIdleWorkers in worker.go.
• Context cancellation propagates through goListenToContext, enabling graceful shutdown.

Frequently Asked Questions

How do I change the number of worker goroutines after starting a worker?

Use the TunePool(int) method available on the worker instance. This updates the internal pool size in internal/pool/pool.go immediately without requiring a restart. Pass the desired concurrency level as an integer; the method returns an error if the pool cannot be resized.

What happens if I do not set an idle worker expiry duration?

When idleWorkerExpiryDuration is unset (zero value), VarMQ retains exactly one idle worker indefinitely regardless of the minIdleWorkerRatio setting. The background cleanup routine goRemoveIdleWorkers only activates when you configure a duration greater than zero, allowing the pool to shrink below the minimum idle threshold after the specified timeout.

Can different workers use different queue selection strategies?

Yes. While you can set a global default with DefaultStrategy, each worker can override this by passing WithStrategy(varmq.Strategy) to its constructor. The queueManager in queue_manager.go respects the worker-specific strategy when calling next() to select jobs from bound queues.

How does VarMQ handle context cancellation?

When you provide a context via WithContext or DefaultCtx, the worker spawns goListenToContext which blocks on ctx.Done(). Upon cancellation, this goroutine automatically invokes worker.Stop(), propagating the cancellation signal to all processing goroutines and ensuring graceful shutdown of the entire worker pool.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}