How to Create a New Worker in VarMQ: Complete Guide with Examples

Create a new worker in VarMQ by importing github.com/goptics/varmq and calling one of the generic factory functions—NewWorker, NewErrWorker, or NewResultWorker—passing your business logic and optional configuration, then bind the returned instance to a queue using BindQueue() or related binder methods.

VarMQ is a high-performance, generic job queue library for Go that manages worker pools with automatic concurrency scaling and graceful shutdown semantics. This guide explains how to create a new worker in VarMQ using the public API surface defined in main.go, including factory selection, configuration options, and queue binding patterns based on the goptics/varmq source code.

Understanding VarMQ Worker Types

VarMQ provides three distinct factory functions in main.go (lines 25–99) to accommodate different return signatures. Choosing the correct type ensures type-safe job processing without unnecessary boilerplate.

NewWorker for Fire-and-Forget Jobs

Use NewWorker when your job performs side effects without returning values or errors. The signature accepts a function of type func(Job[T]) where T is your payload type.

According to the source in main.go lines 25–42, this factory initializes a worker that wraps your function with panic recovery and metrics collection defined in worker.go lines 24–40.

NewErrWorker for Error-Only Returns

Use NewErrWorker when jobs return only an error to indicate success or failure. This is common for validation tasks or I/O operations where the result is binary.

The factory definition resides in main.go lines 43–63, creating a worker that automatically handles error propagation to the job result channel.

NewResultWorker for Result and Error Returns

Use NewResultWorker when jobs compute a value that callers need to retrieve. The worker function signature is func(Job[T]) (R, error), returning both a result of type R and an error.

As implemented in main.go lines 84–99, this factory creates the most complex worker type, managing bidirectional communication between the job submitter and the worker pool.

Step-by-Step Guide to Creating a VarMQ Worker

Creating a production-ready worker involves five distinct steps, from package import to queue binding.

Import the VarMQ Package

Begin by adding the dependency to your module:

go get github.com/goptics/varmq

Then import the package in your Go file:

import "github.com/goptics/varmq"

Select the Appropriate Factory Function

Choose the factory that matches your return requirements:

• Side effects only: varmq.NewWorker(processFunc)
• Error indication: varmq.NewErrWorker(validateFunc)
• Result + Error: varmq.NewResultWorker(computeFunc)

Each factory returns a worker instance that implements the IWorkerBinder interface defined in worker_binder.go (generated by newQueues), exposing binding methods at lines 1–30.

Configure Concurrency and Options

Pass optional configuration arguments as variadic parameters after your function. The loadConfigs function in config.go (lines 1–70) parses these settings:

// Create worker with 8 concurrent goroutines
worker := varmq.NewWorker(processFunc, 8)

// Create worker with custom idle-worker ratio
worker := varmq.NewWorker(processFunc, 8, 0.5) // 50% idle workers

Valid configuration parameters include:

• Concurrency: Integer specifying the number of worker goroutines (default: CPU count)
• IdleWorkerRatio: Float between 0 and 1 determining idle worker pool size
• Context: context.Context for graceful shutdown signaling

Bind to a Queue Type

The worker factory returns a binder implementing IWorkerBinder. Invoke the appropriate binding method based on your queue requirements:

// Standard FIFO queue
q := worker.BindQueue()

// Priority queue (jobs processed by priority value)
pq := worker.BindPriorityQueue()

// Persistent queue (survives process restarts)
dq := worker.BindPersistentQueue("queue-name")

// Distributed queue (for multi-instance deployments)
distQ := worker.BindDistributedQueue("shared-queue-name")

Each binding method is defined in worker_binder.go and connects your worker to the queue management system in queue_manager.go.

Practical Code Examples

Basic Fire-and-Forget Worker

This example demonstrates the simplest worker creation pattern using NewWorker and BindQueue, corresponding to the implementation in main.go lines 25–42:

package main

import (
	"fmt"
	"github.com/goptics/varmq"
)

func main() {
	// Create worker with string payload type
	worker := varmq.NewWorker(func(j varmq.Job[string]) {
		fmt.Println("Processing:", j.Data())
	})

	// Bind to standard FIFO queue
	q := worker.BindQueue()

	// Enqueue jobs (non-blocking)
	q.Add("hello")
	q.Add("world")

	// Wait for completion
	worker.WaitUntilFinished()
}

Error-Handling Worker Pattern

Use NewErrWorker for jobs that may fail but don't return computed values. The worker function receives Job[T] and returns only error:

worker := varmq.NewErrWorker(func(j varmq.Job[string]) error {
	data := j.Data()
	if data == "" {
		return fmt.Errorf("empty payload")
	}
	// Process valid data...
	return nil
})

q := worker.BindQueue()
q.Add("valid-data")

Result-Returning Worker with Concurrency

This example uses NewResultWorker (defined in main.go lines 84–99) to compute values and retrieve them asynchronously. It also demonstrates runtime concurrency configuration:

worker := varmq.NewResultWorker(func(j varmq.Job[int]) (int, error) {
	// Simulate computation
	time.Sleep(200 * time.Millisecond)
	return j.Data() * 2, nil
}, 4) // 4 concurrent workers

q := worker.BindQueue()

// Submit jobs and retrieve results
for i := 1; i <= 5; i++ {
	job, _ := q.Add(i)
	go func(j varmq.ResultJob[int]) {
		if res, err := j.Result(); err == nil {
			fmt.Printf("Result for %d: %d\n", j.Data(), res)
		}
	}(job)
}

worker.WaitUntilFinished()

Runtime Concurrency Tuning

VarMQ supports dynamic adjustment of worker pool size using the TunePool method (implemented in worker.go lines 82–124). This allows scaling workers based on load:

worker := varmq.NewWorker(processFunc, 5)

// Later, increase concurrency to 10
worker.TunePool(10)

// Check current idle workers
idleCount := worker.NumIdleWorkers()

Key Source Files and Architecture

Understanding the internal structure helps debug worker behavior:

• main.go: Contains the public factory functions (NewWorker, NewErrWorker, NewResultWorker) and helper constructors (Func, ErrFunc, ResultFunc) at lines 25–99.
• worker.go: Implements the core worker logic including pool management (initPoolNode), event loops (goEventLoop), idle worker cleanup (goRemoveIdleWorkers), and the TunePool method at lines 82–124 and 150–210.
• worker_binder.go: Generated by newQueues, defines the IWorkerBinder interface with queue binding methods (BindQueue, BindPriorityQueue, BindPersistentQueue, BindDistributedQueue) at lines 1–30.
• config.go: Handles configuration parsing via loadConfigs at lines 1–70, processing concurrency settings and idle-worker ratios.
• queue_manager.go: Orchestrates multiple queue types and provides the next() method for job fetching.

Summary

• Choose the right factory: Use NewWorker for side effects, NewErrWorker for error-only returns, and NewResultWorker when you need to retrieve computed values.
• Configure concurrency: Pass integer arguments to set worker pool size, or use TunePool at runtime to adjust capacity dynamically as shown in worker.go.
• Bind to appropriate queues: The IWorkerBinder interface provides BindQueue, BindPriorityQueue, BindPersistentQueue, and BindDistributedQueue for different persistence and ordering requirements.
• Leverage built-in helpers: The Func() helper simplifies fire-and-forget function execution, though it cannot be used with persistent or distributed queues due to serialization limitations.
• Access source internals: Key logic resides in main.go (factories), worker.go (pool management), and worker_binder.go (queue binding).

Frequently Asked Questions

What is the difference between NewWorker, NewErrWorker, and NewResultWorker?

NewWorker creates workers that perform side effects without returning values, accepting functions of type func(Job[T]). NewErrWorker is designed for operations that return only an error to indicate success or failure. NewResultWorker handles computations that return both a result and an error, enabling callers to retrieve processed values via the Result() method on the job object.

How do I configure the number of concurrent workers when creating a new worker in VarMQ?

Pass the desired concurrency level as an integer argument immediately after your worker function when calling any factory. For example: varmq.NewWorker(processFunc, 8) creates a pool with eight concurrent goroutines. You can also adjust this dynamically after creation using the TunePool() method, which updates the atomic concurrency value and triggers the event loop as implemented in worker.go lines 82–124.

Can I use the Func() helper with persistent or distributed queues?

No, the Func() helper is designed only for in-memory, non-persistent queues. This helper converts standard Go functions into VarMQ job handlers, but because it accepts func() arguments that cannot be serialized, it is incompatible with BindPersistentQueue() or BindDistributedQueue(). For persistent or distributed scenarios, use NewWorker with serializable payload types.

What queue binding options are available after creating a worker?

The worker factory returns an object implementing the IWorkerBinder interface (defined in worker_binder.go), which provides four binding methods: BindQueue() for standard FIFO processing, BindPriorityQueue() for priority-ordered execution, BindPersistentQueue() for disk-backed queues that survive restarts, and BindDistributedQueue() for multi-instance deployments requiring shared queue state.

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