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

> Learn how to create a new worker in VarMQ using the goptics/varmq repository. This guide provides clear examples and steps to integrate your business logic seamlessly.

- Repository: [Goptics/varmq](https://github.com/goptics/varmq)
- Tags: how-to-guide
- Published: 2026-03-02

---

**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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/main.go) lines 25–42, this factory initializes a worker that wraps your function with panic recovery and metrics collection defined in [`worker.go`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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:

```bash
go get github.com/goptics/varmq

```

Then import the package in your Go file:

```go
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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/config.go) (lines 1–70) parses these settings:

```go
// 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:

```go
// 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`](https://github.com/goptics/varmq/blob/main/worker_binder.go) and connects your worker to the queue management system in [`queue_manager.go`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/main.go) lines 25–42:

```go
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`:

```go
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`](https://github.com/goptics/varmq/blob/main/main.go) lines 84–99) to compute values and retrieve them asynchronously. It also demonstrates runtime concurrency configuration:

```go
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`](https://github.com/goptics/varmq/blob/main/worker.go) lines 82–124). This allows scaling workers based on load:

```go
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`](https://github.com/goptics/varmq/blob/main/main.go)**: Contains the public factory functions (`NewWorker`, `NewErrWorker`, `NewResultWorker`) and helper constructors (`Func`, `ErrFunc`, `ResultFunc`) at lines 25–99.
- **[`worker.go`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/worker_binder.go)**: Generated by `newQueues`, defines the `IWorkerBinder` interface with queue binding methods (`BindQueue`, `BindPriorityQueue`, `BindPersistentQueue`, `BindDistributedQueue`) at lines 1–30.
- **[`config.go`](https://github.com/goptics/varmq/blob/main/config.go)**: Handles configuration parsing via `loadConfigs` at lines 1–70, processing concurrency settings and idle-worker ratios.
- **[`queue_manager.go`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/main.go) (factories), [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go) (pool management), and [`worker_binder.go`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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.