Worker pools and job queues for Go

JOB 02Scheduling

Order, fairness and fan-out, each in one line.

Each scene shows the order varmq runs the jobs in, with the call that makes it. Typed results and errors, job IDs and deduplication, context cancellation, metrics, a REST API and Prometheus are in the guides.

// runs: alert, report, digest

Priority queues

Give a job a number. The lowest number runs first, and equal numbers keep the order you added them.

Queues guide

// runs: a1, r1, e1, a2, e2, e3

Several queues, one worker

Bind many queues to one worker. A strategy decides which queue gives the next job: Priority, RoundRobin, MaxLen or MinLen.

Strategies guide

// email and stock each run order 7

Topics

Publish once and every subscribed worker gets its own job. A queue gives each job to one worker.

Topics guide

JOB 03Where jobs livesame worker, three homes

Same worker. You choose where the jobs live.

Press Restart process. A queue in memory ends with the process. A queue in Redis or SQLite keeps the waiting jobs.

In memorySQLiteRedis

Restart processAdd 5 jobs

Built in. Plain queues, priority queues and topics live in your process. They have nothing to set up, and they end when the process ends.

main.goQueues guide

queue := worker.BindQueue()

// or, ordered by a number:
pq := worker.BindPriorityQueue()

Use it when a lost job is fine, or when you can add it again.

sqliteq keeps jobs in local SQLite files, so a single process keeps them across a restart with no server to run.

main.goSQLite guide

sdb, err := sqliteq.New("data")
if err != nil {
	panic(err)
}
defer sdb.Close()

pq, err := sdb.NewQueue("jobs")
if err != nil {
	panic(err)
}
queue := worker.WithPersistentQueue(pq)

A job that was running when the process died can run again. Write jobs so a second run is safe.

redisq keeps jobs in Redis. A persistent queue lets one process keep its jobs across a restart. A distributed queue lets several processes consume one queue, and each job goes to one of them.

main.goRedis guide

rdb := redisq.New("redis://localhost:6375")
defer rdb.Close()

pq := rdb.NewQueue("persistent_queue")
queue := worker.WithPersistentQueue(pq)

// several processes: rdb.NewDistributedQueue(name)
// with worker.WithDistributedQueue

A job that was running when the process died can run again. Write jobs so a second run is safe.

JOB 04Measured in publicraw output published

The numbers, and where they do not favor VarMQ.

Each lane is one library on the same harness, same machine and same run. The longest bar is the highest value. The last column is the lane's value divided by VarMQ's.

Worker poolRedis queuesSQLite queues

Replay

Mean tasks per second over 15 rows: 3 task types × 5 producer shapes (1 to 1,000,000 producers), 50K workers, 1M tasks per run. Averaged over GOMAXPROCS 24, 8 and 4. Memory is bytes allocated per run of 1M tasks, from the same rows.

Throughput

tasks per second · higher is better

LibraryRate× VarMQ

1. VarMQ4.89M1.00×
2. Buffered channel †2.73M0.56×
3. errgroup (opens in a new tab)2.29M0.47×
4. ants (opens in a new tab)2.22M0.45×
5. gopool (opens in a new tab)1.92M0.39×
6. Channel †1.90M0.39×
7. conc (opens in a new tab)1.56M0.32×
8. workerpool (opens in a new tab)1.35M0.28×
9. pond (opens in a new tab)0.60M0.12×

Memory

MB allocated per 1M tasks · lower is better

LibraryAllocated× VarMQ

1. Channel †14.5 MB0.13×
2. conc (opens in a new tab)15.0 MB0.13×
3. ants (opens in a new tab)15.4 MB0.13×
4. Buffered channel †23.2 MB0.20×
5. workerpool (opens in a new tab)28.2 MB0.25×
6. gopool (opens in a new tab)52.0 MB0.45×
7. errgroup (opens in a new tab)110.7 MB0.96×
8. VarMQ115.0 MB1.00×
9. pond (opens in a new tab)317.8 MB2.76×

† Baselines, not libraries: N goroutines reading one channel. The buffered one has a channel that holds all 1M tasks.

100,000 no-op jobs through one library instance with 100 workers, Redis 7.2 on one pinned core. VarMQ runs on redisq. Median time from enqueue to handler, one job at a time: VarMQ 0.058 ms, the lowest of the run. taskq 0.062 ms, gocraft/work 0.082 ms, rmq 1.25 ms, Asynq 998 ms.

100 producers1 producer

100 producers enqueue at once

jobs per second · higher is better

LibraryRate× VarMQ

1. VarMQ59,8981.00×
2. gocraft/work (opens in a new tab)44,7750.75×
3. taskq (opens in a new tab)34,4990.58×
4. rmq (opens in a new tab)28,5210.48×
5. Asynq (opens in a new tab)17,7440.30×

One producer enqueues 100,000 jobs

jobs per second · higher is better

LibraryRate× VarMQ

1. rmq (opens in a new tab)35,7081.72×
2. gocraft/work (opens in a new tab)25,7311.24×
3. VarMQ20,8121.00×
4. taskq (opens in a new tab)18,4480.89×
5. Asynq (opens in a new tab)15,0960.73×

Drain 20,000 jobs added by 4 producers, WAL checkpoint every 1,000 pages. VarMQ runs on sqliteq. In this run VarMQ was the fastest subject in all 8 scenario and checkpoint groups.

Drain one queue

jobs per second · higher is better

LibraryRate× VarMQ

1. VarMQ14,7251.00×
2. River (opens in a new tab)3,2300.22×
3. goqite (opens in a new tab)2,9330.20×
4. backlite (opens in a new tab)1,1190.08×

What the same run says against VarMQ

• Memory. VarMQ allocates 115 MB per 1M tasks. ants allocates 15 MB, which is 7.5× less. Add never blocks, so waiting tasks sit in an unbounded queue.
• One worker. On the 15 one-worker Loop10 rows, a plain buffered channel is fastest in all 15.
• 4 CPUs. The buffered channel wins 21 of 55 rows and VarMQ wins 19. At 24 CPUs VarMQ wins 39 and the channel 12.
• Redis, one producer. rmq leads 12 of the 13 single-producer rows. VarMQ leads the enqueue-only row, and leads the 100-producer rows with one instance at 100 and 1,000 workers.

Run of 2026-10-25 · varmq v1.10.0 · Intel Core i7-13700 (24 threads), linux/amd64, Go 1.26 · Absolute numbers vary between runs, so compare ratios within one run. Figures come from the raw output in github.com/goptics/benchmarks; charts and the harness are at benchmarks.goptics.org.

JOB 05Programseach one runs as it is

Each tab is a full program.

Press Run to run it in the Go Playground, or edit it first.

Quick startPriorityRound robinTopicsResults

A worker with the default concurrency of 1 runs two jobs, one after the other. Read it in the docs

quick-start/main.go

RunCopyEditPlayground ↗

9 lines: package and imports

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)
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

Output

The worker starts after the three jobs are added, so the lowest number runs first. Read it in the docs

queues-priority/main.go

RunCopyEditPlayground ↗

8 lines: package and imports

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func main() {
	// Not started, so all three jobs wait in the queue together.
	worker := varmq.NewWorker(
		func(j varmq.Job[string]) {
			fmt.Println("running", j.Data())
		},
		varmq.WithAutoRun(false),
	)

	queue := worker.BindPriorityQueue()

	queue.Add("report", 10)
	queue.Add("alert", 1)
	queue.Add("digest", 10)

	worker.Start()
	worker.WaitUntilIdle()
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

Output

One worker takes jobs from three queues in turn. Read it in the docs

strategies-round-robin/main.go

RunCopyEditPlayground ↗

8 lines: package and imports

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(
		func(j varmq.Job[string]) {
			fmt.Println(j.Data())
		},
		varmq.WithStrategy(varmq.RoundRobin),
		varmq.WithAutoRun(false),
	)

	alerts := worker.BindQueue()
	reports := worker.BindQueue()
	emails := worker.BindQueue()

	alerts.Add("alert 1")
	alerts.Add("alert 2")
	reports.Add("report 1")
	emails.Add("email 1")
	emails.Add("email 2")
	emails.Add("email 3")

	worker.Start()
	worker.WaitUntilIdle()
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

Output

Two workers subscribe to one topic, and each gets every order. Read it in the docs

topic-subscribe/main.go

RunCopyEditPlayground ↗

9 lines: package and imports

package main

import (
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	topic := varmq.NewTopic[int]()

	email := varmq.NewWorker(func(j varmq.Job[int]) {
		time.Sleep(500 * time.Millisecond)
		fmt.Println("email for order", j.Data())
	})

	stock := varmq.NewWorker(func(j varmq.Job[int]) {
		time.Sleep(10 * time.Millisecond)
		fmt.Println("stock for order", j.Data())
	}, 2)

	email.WithTopic(topic)
	stock.WithTopic(topic)

	for i := range 4 {
		topic.Publish(i) // both workers get their own job for this order
	}

	email.WaitUntilIdle()
	stock.WaitUntilIdle()
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

Output

A result worker returns a value and an error, which you read from the job. Read it in the docs

worker-types-result/main.go

RunCopyEditPlayground ↗

9 lines: package and imports

package main

import (
	"fmt"
	"strconv"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewResultWorker(func(j varmq.Job[string]) (int, error) {
		return strconv.Atoi(j.Data())
	})
	queue := worker.BindQueue()

	for _, s := range []string{"42", "forty-two"} {
		job, _ := queue.Add(s)
		n, err := job.Result()
		fmt.Printf("%q: %d, %v\n", s, n, err)
	}
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

Output

JOB 06Indexwhere to read next

Where to read next

• 01StartInstall varmq and run your first worker.→
• 02ConceptsWorkers, queues, jobs and the lifecycle.→
• 03GuidesConcurrency, errors, waiting, metrics and more.→
• 04PersistenceRedis and SQLite adapters, and writing your own.→
• 05OperationsControl a running worker over HTTP. Export Prometheus metrics.→
• 06ComparisonFeatures side by side with other Go worker pools.→
• 07API referenceEvery type and function, on pkg.go.dev.→

tear here

Run your first worker.

varmq+ Redis+ SQLite

go get github.com/goptics/varmqCopy

go get github.com/goptics/varmq github.com/goptics/redisqCopy

go get github.com/goptics/varmq github.com/goptics/sqliteqCopy

Quick startRead the docs