goptics/varmq: A Simplest in-memory and Distributed Storage-Agnostic Message Queue for Your Concurrent Go Program

Features

• High performance: lock-free appends and batched dispatch. See benchmarks.
• Queue types: standard, priority, persistent (survives restarts) and distributed (shared across instances).
• Worker types: NewWorker (fire and forget), NewErrWorker (error only), NewResultWorker (result and error).
• Concurrency control: resize the pool at runtime, pause, resume, restart.
• Multi queue binding: one worker, many queues, a pluggable selection strategy.
• Topic fan-out: publish once, every subscribed worker gets its own job.
• REST API: inspect and control workers over HTTP.
• Extensible: plug in your own storage by implementing IPersistentQueue or IDistributedQueue.

Quick Start

go get github.com/goptics/varmq

package main
import (
	"fmt"
	"time"
"github.com/goptics/varmq"
)
func main() {
	worker := varmq.NewWorker(func(j varmq.Job[int]) {
		fmt.Printf("Processing %d\n", j.Data())
		time.Sleep(1 * time.Second)
	}) // default concurrency is 1
	defer worker.WaitUntilIdle()
	queue := worker.BindQueue()
queue.Add(1)
	queue.Add(2)
}

↗️ Run it on Playground

Usage

Runnable programs for most sections live in examples.

Concurrency

worker := varmq.NewWorker(func(j varmq.Job[int]) {
	fmt.Printf("Processing %d\n", j.Data())
	time.Sleep(500 * time.Millisecond)
}, 10) // concurrency 10, or 0 for one per CPU
defer worker.WaitUntilIdle()
queue := worker.BindQueue()
for i := range 100 {
	queue.Add(i)
}

↗️ Run it on Playground

[!Note]

Concurrency is a limit, not a fixed pool size: goroutines start while jobs wait and exit as the queue drains. Tune it with varmq.WithMinIdleWorkerRatio and varmq.WithIdleWorkerExpiryDuration.

Priority Queue

Lower number = higher priority. A higher-priority job added later is dispatched first.

queue := worker.BindPriorityQueue()
for i := range 10 {
	queue.Add(i, i%2) // even jobs first
}

↗️ Run it on Playground

!VarMQ priority queue with one worker

Deduplicate by Job ID

An in-memory queue holds at most one waiting job per id. The id is free again once a worker takes the job or Purge drops it. Jobs without an id are never deduplicated.

queue.Add(userID, varmq.WithJobId("refresh:"+userID)) // queued
queue.Add(userID, varmq.WithJobId("refresh:"+userID)) // nil, false: one is already waiting

Persistent and Distributed Queues

Plug in a storage adapter to keep jobs across restarts or share them between instances:

queue := worker.WithPersistentQueue(persistentAdapter)   // any IPersistentQueue
queue := worker.WithDistributedQueue(distributedAdapter) // any IDistributedQueue

Examples: SQLite persistent, Redis persistent, Redis distributed.

When the adapter deduplicates pushes (IDedupQueue, e.g. redisq), give each job a unique id such as a UUID. A retried Add with the same id is then stored once:

id := uuid.NewString() // once, so a retry carries the same id
for !queue.Add(data, varmq.WithJobId(id)) {
	// back off and retry
}

Built-in adapters

• ⚡ Redis: redisq
• 🗃️ SQLite: sqliteq
• 🐘 PostgreSQL: 🔄 Upcoming

Multi Queue Binds

Bind several queues to one worker. varmq.WithStrategy picks which queue the next job comes from:

1. Priority (default): the highest-priority queue with pending jobs
2. RoundRobin: take turns across queues
3. MaxLen: the longest queue
4. MinLen: the shortest queue

worker := varmq.NewWorker(func(j varmq.Job[string]) {
	fmt.Println("Processing:", j.Data())
	time.Sleep(500 * time.Millisecond)
})
defer worker.WaitUntilIdle()
// queue priority follows bind order; override with varmq.WithQueuePriority
q1 := worker.BindQueue()         // highest
q2 := worker.BindQueue()         // medium
pq := worker.BindPriorityQueue() // lowest
for i := range 15 {
	q2.Add(fmt.Sprintf("Task queue-2 %d", i))
}
for i := range 10 {
	pq.Add(fmt.Sprintf("Task priority-queue %d", i), i%2)
}
for i := range 10 {
	q1.Add(fmt.Sprintf("Task queue-1 %d", i))
}

↗️ Run it on Playground

With varmq.RoundRobin, empty queues are skipped:

!VarMQ multiple queues with round-robin selection

Topic (Fan-out)

One producer, many workers: publish once and every subscribed worker gets its own job. Publish returns false when no worker is subscribed, the topic is closed, or it is full.

| | Queue | Topic | |---|---|---| | Who gets a job | each job is processed once, by one worker | every subscribed worker gets its own job | | Storage | the job is removed once a worker takes it | the payload is stored once; each worker reads it with its own cursor, and it's freed after every worker has read it | | Producer | queue.Add(x) | topic.Publish(x) | | Late joiner | sees jobs still waiting in the queue | sees only what's published after it subscribed |

topic := varmq.NewTopic[int]()
email := varmq.NewWorker(func(j varmq.Job[int]) {
	fmt.Printf("Emailing order %d\n", j.Data())
	time.Sleep(1 * time.Second)
})
stock := varmq.NewWorker(func(j varmq.Job[int]) {
	fmt.Printf("Updating stock for order %d\n", j.Data())
	time.Sleep(300 * time.Millisecond)
}, 2)
defer email.WaitUntilIdle()
defer stock.WaitUntilIdle()
email.WithTopic(topic)
stock.WithTopic(topic)
for order := range 4 {
	topic.Publish(order) // both workers get all 4 orders
}

↗️ Run it on Playground

!VarMQ topic fan-out with two subscribed workers

Result and Error Worker

NewResultWorker returns a result and an error per job. NewErrWorker returns only the error.

worker := varmq.NewResultWorker(func(j varmq.Job[string]) (int, error) {
	if j.Data() == "error" {
		return 0, errors.New("error occurred")
	}
	return len(j.Data()), nil
})
defer worker.WaitUntilIdle()
queue := worker.BindQueue()
// Add returns at once; Result blocks until that job is done
if job, ok := queue.Add("The length of this string is 31"); ok {
	result, err := job.Result()
	fmt.Println("Result:", result, err) // Result: 31 <nil>
}
if job, ok := queue.Add("error"); ok {
	_, err := job.Result()
	fmt.Println("Error:", err) // Error: error occurred
}

↗️ Run it on Playground

Function Helpers

Submit functions directly, like Submit() in pond or ants. Use Func() with NewWorker, ErrFunc() with NewErrWorker and ResultFunc[R]() with NewResultWorker.

worker := varmq.NewWorker(varmq.Func(), 10)
defer worker.WaitUntilIdle()
queue := worker.BindQueue()
for i := range 100 {
	queue.Add(func() {
		time.Sleep(500 * time.Millisecond)
		fmt.Println("Processing", i)
	})
}

↗️ Run it on Playground

[!Important]

Functions can't be serialized, so function helpers don't work with persistent or distributed queues.

Lifecycle and Errors

Pause, Resume, TunePool and Stop return at once. Restart waits for the worker to stop before starting it again, and the Wait* methods block until their condition is met. Running jobs always finish.

worker := varmq.NewWorker(process, 4,
	varmq.WithContext(ctx), // cancelling ctx stops the worker
	varmq.WithErrHandler(func(w varmq.Worker, err error) {
		log.Println("varmq:", err) // dequeue, parse and other internal errors
	}),
)
worker.Pause()     // stop taking new jobs; unstarted jobs stay queued
worker.Resume()    // carry on
worker.TunePool(8) // change concurrency on the fly
worker.Restart()   // stop and start again, keeping pending jobs
worker.Stop()      // graceful stop
worker.Wait()          // block until no job is running or pending
worker.WaitUntilIdle() // block until the worker is Idle (never returns while paused or stopped)

Status() reports the state (Running, Idle, Paused, Stopped, ...). PauseAndWait, StopAndWait, WaitUntilPaused and WaitUntilStopped block until a state is reached.

REST API

Mount the handler on any http.ServeMux:

mux := http.NewServeMux()
mux.Handle("/varmq/", varmq.Handler("/varmq"))
http.ListenAndServe(":8080", mux)

| Method | Path | Description | |----------|-------------------------------------------|------------------------------| | GET | /health | Health check | | GET | /workers | List all workers (summary) | | GET | /workers/{name} | Get worker details + metrics | | PATCH | /workers/{name}/actions/{action} | Pause, resume, stop, restart | | PATCH | /workers/{name}/config/concurrency/{n} | Adjust concurrency |

[!Warning]

The REST API is unauthenticated by design. Protect it with middleware, network isolation, or a reverse proxy before exposing to untrusted networks.

How It Works

Event loop

A single event loop sleeps until something signals it: Add() putting a job in the queue, or a worker finishing one. On each signal it hands jobs to workers while the queue has work and a worker slot is free, then goes back to waiting.

!VarMQ event loop with one worker

Worker pool

Every Add (and every job in AddAll) tries to wake the event loop, but the signal channel holds only one pending signal, so the extra signals are simply dropped. One wake-up is enough: the loop keeps dispatching jobs back to back while the queue has work and a worker slot is free, and it hands each worker its share of the backlog in one go (a batch) rather than one job per wake-up.

!VarMQ worker pool with concurrency 4

Batched dispatch

A hand-off costs the event loop the same whether a job takes a nanosecond or a minute, so with many short jobs the loop, not the workers, becomes the bottleneck. To avoid that, a worker can take several jobs at once (a batch) and run them in order.

Batch size is the smallest of: an even split of the waiting jobs across workers, enough jobs to keep hand-offs about 32 µs apart (sized from the last finished batch; one job until a job has been measured), and 32. With enough jobs waiting:

| Setup | Jobs per hand-off | |---|---| | 10 workers, 3 s jobs | 1, so jobs start in queue order | | 24 workers, 1 ms jobs | 1 | | 3,000 workers, 10 ms jobs | 10 | | 10,000+ workers, 10 ms jobs | 32 | | Concurrency 1, any jobs | up to 32: one worker runs jobs in order anyway |

Below, 8 cheap jobs with concurrency 2: the first worker takes 4, a second worker is started for 2, and the last 2 go out one at a time as workers free up.

!VarMQ batched dispatch

• Jobs waiting in a batch stay Queued: NumPending counts them, NumProcessing does not, and you can still close them.
• On Stop or Pause each worker finishes its running job and parks the rest of its batch. NumPending counts parked jobs but the queue's Len does not. They run first, in order, on Resume or Restart, and Purge closes them.
• Capacity counts only jobs still in the queue, so up to 32 more per worker can be pending.

Comparison and Benchmarks

VarMQ is compared with ants, pond v2, conc, workerpool, errgroup, gopool and two hand-rolled channel pools as a worker pool, and with Redis- and SQLite-backed job queues as a persistent queue, in the benchmark repository.

Feature comparison

| Feature | VarMQ | ants | pond | conc | workerpool | errgroup | gopool | | ---------------------------------------- | :---: | :--: | :-----: | :--: | :--------: | :------: | :----: | | Generic (type-safe) API | ✅ | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ | | Tasks return results and errors | ✅ | ❌ | ✅ | ✅ | ❌ | ❌² | ❌ | | Task groups (submit many, wait once) | ✅ | ❌ | ✅ | ✅ | ❌ | ✅ | ❌ | | Priority queue | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | | Persistent queues (survive restarts) | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | | Distributed queues (across instances) | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | | Several queues bound to one pool | ✅ | ❌ | ✅⁴ | ❌ | ❌ | ❌ | ❌ | | Topic fan-out (publish once, many pools) | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | | Bounded queue, reject when full | ✅ | ✅ | ✅ | ❌ | ❌ | ❌³ | ❌ | | Pause and resume | ✅ | ❌ | ❌ | ❌ | ✅ | ❌ | ❌ | | Restart after stop | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | | Resize pool at runtime | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ | ✅ | | Cancel work via context | ✅ | ❌ | ✅ | ✅ | ❌ | ✅ | ❌ | | Panic recovery | ✅ | ✅ | ✅ | ❌¹ | ❌ | ❌ | ✅ | | Task counters (submitted, failed, ...) | ✅ | ❌ | ✅ | ❌ | ❌ | ❌ | ❌ | | Built-in REST API | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |

_{Checked against each package's source, October 2026. ¹ conc re-raises a task's panic from Wait by design. ² errgroup returns the first error, no results. ³ errgroup has no queue; TryGo only refuses a task while every slot is busy. ⁴ pond subpools: each has its own queue and runs on the parent pool's workers.}

Performance

Run of 2026-10-25 on an i7-13700 (24 threads), Go 1.26, varmq v1.10.0. Raw results and harness in goptics/benchmarks, charts at benchmarks.goptics.org.

Worker pool

Every pool gets 50K workers and runs three kinds of task (sleep 10 ms, a short CPU loop, and a mix of both), 5M tasks of each, sent by anywhere from 1 to 1M producers. Live charts: ↗️ 24 CPUs · ↗️ 8 CPUs · ↗️ 4 CPUs.

Time to run all 15M tasks per CPU count and memory per 1M tasks, lower is better; throughput in million tasks per second, averaged over every run, higher is better.

| Pool | 4 CPUs (s) | 8 CPUs (s) | 24 CPUs (s) | Throughput (M/s) | Memory (MB) | | ---------------- | ---------: | ---------: | ----------: | ---------------: | ----------: | | VarMQ | 5.07 | 3.54 | 3.85 | 4.89 | 115 | | BufferedPool⁵ | 5.43 | 5.41 | 8.49 | 2.73 | 23 | | errgroup | 6.38 | 5.69 | 9.23 | 2.29 | 111 | | ants | 6.74 | 6.51 | 8.00 | 2.22 | 15 | | gopool | 7.13 | 7.48 | 11.00 | 1.92 | 52 | | GoroutinePool⁵ | 7.24 | 6.96 | 12.85 | 1.90 | 14 | | conc | 8.45 | 8.61 | 17.38 | 1.56 | 15 | | workerpool | 10.20 | 11.58 | 16.42 | 1.35 | 28 | | pond v2 | 20.67 | 22.23 | 42.38 | 0.60 | 318 |

Why more memory: Add never blocks, so VarMQ queues every task as a job (ID, status, result handle) until a worker takes it; errgroup starts a goroutine per task. ants, conc and the unbuffered pool make the producer wait instead, so far fewer tasks are held at once.

The full matrix (100K, 300K and single-worker pools) is on the live charts; the harness is in pool.

_{⁵ Hand-rolled: fixed workers reading a channel, unbuffered (GoroutinePool) or with 1M slots (BufferedPool).}

Redis-backed queues

SQLite-backed queues

Micro-benchmarks

goos: linux
goarch: amd64
pkg: github.com/goptics/varmq
cpu: 13th Gen Intel(R) Core(TM) i7-13700

API Reference

For the complete package documentation, types, and method signatures, browse the GoDoc.

Star History

make init  # install lefthook git hooks and tidy modules
make test  # go test -race ./...
make bench # micro-benchmarks

The Redis examples need a local Redis: docker compose -f examples/compose.yml up -d. Commit messages follow Conventional Commits (feat:, fix:, docs:, ...).

This project has a Code of Conduct. By participating, you agree to its terms.

License

MIT