Introduction

varmq is a typed worker pool and message queue for Go. You give a worker a function and a concurrency limit, bind one or more queues to it, and add jobs to those queues. The worker runs each job through your function, never more at once than the limit allows.

When to use it

Reach for varmq when a program has work to run in the background and you want a say in how much of it runs at once:

• Bounded concurrency with typed jobs. A Job[T] carries your data type, and a job can hand back an error (NewErrWorker) or a result and an error (NewResultWorker).
• Several sources, one pool. One worker can take jobs from plain queues, priority queues and topics at the same time.
• Control at runtime. Pause, resume, restart or resize a running worker.
• Jobs that outlive the process. Plug in a storage adapter, such as redisq or sqliteq, to keep jobs across restarts or share a queue between instances.
• Fan-out. Publish to a topic once and every subscribed worker gets its own job. Topics are in-memory only.
• Inspection over HTTP. A Handler lists workers and pauses, resumes or resizes them.

If all you need is to run a fixed set of functions and wait for them, errgroup or a sync.WaitGroup is enough.

The model: a worker is a pool, a queue is a source

varmq splits the work between two things.

A worker is the pool. It owns your function and the concurrency limit. The limit is an upper bound, not a fixed number of goroutines: the worker starts goroutines while jobs are waiting and lets them exit once the queue drains.

A queue is a source of jobs. Each queue is bound to one worker. Add puts a job in the queue and returns at once; the worker’s event loop then hands waiting jobs to free goroutines.

Below, six jobs are added to a worker with concurrency 4. Four start right away, two wait in the queue, and the pool shrinks back to one idle goroutine once the queue is empty.

Worker pool · simulation

loop

asleep

busy

0

queued

0

done

0

A worker pool drains a backlog. Press Add to put jobs in the queue. The jobs arrive as one wake-up for the event loop, which hands them to pool goroutines while a slot is free. The goroutines spawn on demand and the pool shrinks back to one idle goroutine when the queue is empty. Steppers change the concurrency limit and the number of jobs. Play, Pause and Step control the playback.

queue

event loop

asleep

–signal slot, empty

pool · up to 4

done

Press Add to put jobs in the queue.

worker := varmq.NewWorker(process, 4) queue := worker.BindQueue()

Add 6 jobsPauseStepReset

limit

−

4

+

jobs

−

6

+

Each job takes about a second, so every hand-off is one job when the limit is above 1. At limit 1 the one goroutine takes the whole backlog in one batch (batching is below). A model of the library, not a benchmark.

Next: install varmq, then run the quick start in your browser.