VarMQ

A high-performance message queue and pool system for Go that simplifies concurrent task processing using worker pool. Through Go generics, it provides type safety without sacrificing performance.

With VarMQ, you can process messages asynchronously, handle errors properly, store data persistently, and scale across systems using adapters. All through a clean, intuitive API that feels natural to Go developers.

✨ Features

• ⚡ High performance: Optimized for high throughput with minimal overhead, even under heavy load. see benchmarks
• 🛠️ Variants of queue types:
  • Standard queues for in-memory processing
  • Priority queues for importance-based ordering
  • Persistent queues for durability across restarts
  • Distributed queues for processing across multiple systems
• 🧩 Worker abstractions:
  • NewWorker - Fire-and-forget operations (most performant)
  • NewErrWorker - Returns only error (when result isn't needed)
  • NewResultWorker - Returns result and error
• 🚦 Concurrency control: Fine-grained control over worker pool size, dynamic tuning and idle workers management
• 🧬 Multi Queue Binding: Bind multiple queues to a single worker
• 💾 Persistence: Support for durable storage through adapter interfaces
• 🌐 Distribution: Scale processing across multiple instances via adapter interfaces
• 🧩 Extensible: Build your own storage adapters by implementing VarMQ's internal queue interfaces

Quick Start

Installation

go get github.com/goptics/varmq

Basic Usage

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(500 * time.Millisecond)
  }, 10) // with concurrency 10
  defer worker.WaitUntilFinished()
  queue := worker.BindQueue()

  for i := range 100 {
    queue.Add(i)
  }
}

↗️ Run it on Playground

Priority Queue

You can use priority queue to prioritize jobs based on their priority. Lower number = higher priority

// just bind priority queue
queue := worker.BindPriorityQueue()

// add jobs to priority queue
for i := range 10 {
    queue.Add(i, i%2) // prioritize even tasks
}

↗️ Run it on Playground

💡 Highlighted Features

Persistent and Distributed Queues

VarMQ supports both persistent and distributed queue processing through adapter interfaces:

• Persistent Queues: Store jobs durably so they survive program restarts
• Distributed Queues: Process jobs across multiple systems

Usage is simple:

// For persistent queues (with any IPersistentQueue adapter)
queue := worker.WithPersistentQueue(persistentQueueAdapter)

// For distributed queues (with any IDistributedQueue adapter)
queue := worker.WithDistributedQueue(distributedQueueAdapter)

See complete working examples in the examples directory:

• Persistent Queue Example (SQLite)
• Persistent Queue Example (Redis)
• Distributed Queue Example (Redis)

Create your own adapters by implementing the IPersistentQueue or IDistributedQueue interfaces.

Note

Before testing examples, make sure to start the Redis server using docker compose up -d.

Built-in adapters

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

Multi Queue Binds

Bind multiple queues to a single worker, enabling efficient processing of jobs from different sources with configurable strategies. The worker supports three strategies:

1. RoundRobin (default - cycles through queues equally)
2. MaxLen (prioritizes queues with more jobs)
3. MinLen (prioritizes queues with fewer jobs)

worker := varmq.NewWorker(func(j varmq.Job[string]) {
  fmt.Println("Processing:", j.Data())
  time.Sleep(500 * time.Millisecond) // Simulate work
}) // change strategy through using varmq.WithStrategy
defer worker.WaitUntilFinished()

// Bind to a standard queues
q1 := worker.BindQueue()
q2 := worker.BindQueue()
pq := worker.BindPriorityQueue()

for i := range 10 {
  q1.Add(fmt.Sprintf("Task queue 1 %d", i))
}

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) // prioritize even tasks
}

↗️ Run it on Playground

It will process jobs from all queues in a round-robin fashion.

Result and Error Worker

VarMQ provides a NewResultWorker that returns both the result and error for each job processed. This is useful when you need to handle both success and failure cases.

worker := varmq.NewResultWorker(func(j varmq.Job[string]) (int, error) {
 fmt.Println("Processing:", j.Data())
 time.Sleep(500 * time.Millisecond) // Simulate work
 data := j.Data()

 if data == "error" {
  return 0, errors.New("error occurred")
 }

 return len(data), nil
})
defer worker.WaitUntilFinished()
queue := worker.BindQueue()

// Add jobs to the queue (non-blocking)
if job, ok := queue.Add("The length of this string is 31"); ok {
 fmt.Println("Job 1 added to queue.")

 go func() {
  result, _ := job.Result()
  fmt.Println("Result:", result)
 }()
}

if job, ok := queue.Add("error"); ok {
 fmt.Println("Job 2 added to queue.")

 go func() {
  _, err := job.Result()
  fmt.Println("Result:", err)
 }()
}

↗️ Run it on Playground

NewErrWorker is similar to NewResultWorker but it returns only error.

Function Helpers

VarMQ provides helper functions that enable direct function submission similar to the Submit() pattern in other pool packages like Pond or Ants

• Func(): For basic functions with no return values - use with NewWorker
• ErrFunc(): For functions that return errors - use with NewErrWorker
• ResultFunc[R](): For functions that return a result and error - use with NewResultWorker

worker := varmq.NewWorker(varmq.Func(), 10)
defer worker.WaitUntilFinished()

queue := worker.BindQueue()

for i := range 100 {
    queue.Add(func() {
        time.Sleep(500 * time.Millisecond)
        fmt.Println("Processing", i)
    })
}

↗️ Run it on Playground

Important

Function helpers don't support persistence or distribution since functions cannot be serialized.

Benchmarks

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

Add Operation

Command: go test -run=^$ -benchmem -bench '^(BenchmarkAdd)$' -cpu=1

Why use -cpu=1? Since the benchmark doesn’t test with more than 1 concurrent worker, a single CPU is ideal to accurately measure performance.

Worker Type	Queue Type	Time (ns/op)	Memory (B/op)	Allocations (allocs/op)
Worker	Queue	918.6	128	3
	Priority	952.7	144	4
ErrWorker	ErrQueue	1017	305	6
	ErrPriority	1006	320	7
ResultWorker	ResultQueue	1026	353	6
	ResultPriority	1039	368	7

AddAll Operation

Command: go test -run=^$ -benchmem -bench '^(BenchmarkAddAll)$' -cpu=1

Worker Type	Queue Type	Time (ns/op)	Memory (B/op)	Allocations (allocs/op)
Worker	Queue	635,186	146,841	4,002
	Priority	755,276	162,144	5,002
ErrWorker	ErrQueue	673,912	171,090	4,505
	ErrPriority	766,043	186,663	5,505
ResultWorker	ResultQueue	675,420	187,897	4,005
	ResultPriority	777,680	203,263	5,005

Note

AddAll benchmarks use a batch of 1000 items per call. The reported numbers (ns/op, B/op, allocs/op) are totals for the whole batch. For per-item values, divide each by 1000.
e.g. for default Queue, the average time per item is approximately 635ns.

Why is AddAll faster than individual Add calls? Here's what makes the difference:

1. Batch Processing: Uses a single group job to process multiple items, reducing per-item overhead
2. Shared Resources: Utilizes a single result channel for all items in the batch

Charts

Metric	Add Operation	AddAll Operation
Execution Time	Time (ns/op)	Time (ms/op)
Memory Usage	Memory (B/op)	Memory (KB/op)
Allocations	Allocations (allocs/op)	Allocations (allocs/op)

Chart images is been generated using Vizb

Comparison with Other Packages

We conducted comprehensive benchmarking between VarMQ and Pond v2, as both packages provide similar worker pool functionalities. While VarMQ draws inspiration from some of Pond's design patterns, it offers unique advantages in queue management and persistence capabilities.

Key Differences:

• Queue Types: VarMQ provides multiple queue variants (standard, priority, persistent, distributed) vs Pond's single pool type
• Multi-Queue Management: VarMQ supports binding multiple queues to a single worker with configurable strategies (RoundRobin, MaxLen, MinLen)

For detailed performance comparisons and benchmarking results, visit:

• 📊 Benchmark Repository - Complete benchmark suite
• 📈 Interactive Charts - Visual performance comparisons

API Reference

For detailed API documentation, see the API Reference.

The Concurrency Architecture

VarMQ's concurrency model is built around a smart event loop that keeps everything running smoothly.

The event loop continuously monitors for pending jobs in queues and available workers in the pool. When both conditions are met, jobs get distributed to workers instantly. When there's no work to distribute, the system enters a low-power wait state.

Workers operate independently - they process jobs and immediately signal back when they're ready for more work. This triggers the event loop to check for new jobs and distribute them right away.

The system handles worker lifecycle automatically. Idle workers either stay in the pool or get cleaned up based on your configuration, so you never waste resources or run short on capacity.

Star History

Contributing

Contributions are welcome! Please feel free to submit a Pull Request or open an issue.

Please note that this project has a Code of Conduct. By participating in this project, you agree to abide by its terms.