Workers and queues

A worker runs jobs; a queue holds them until the worker has a free slot. Apart from a producer-only distributed queue (NewDistributedQueue), you ask a worker for a queue, and that binds the two.

Binding a queue

BindQueue creates an in-memory, first-in first-out queue, binds it to the worker and returns a Queue[T] handle. You add jobs through the handle; the worker takes them from there.

worker := varmq.NewWorker(process, 2)
queue := worker.BindQueue()
queue.Add("hello")

BindQueue is one of several ways to bind a source to a worker (IWorkerBinder lists them all):

Method	Binds
BindQueue	a new in-memory FIFO queue
BindPriorityQueue	a new in-memory priority queue
WithQueue, WithPriorityQueue	your own IQueue or IPriorityQueue implementation
WithPersistentQueue, WithDistributedQueue	a storage adapter, such as redisq or sqliteq
WithPersistentPriorityQueue, WithDistributedPriorityQueue	a storage adapter’s priority queue
WithTopic	a subscription to a topic, which you publish to instead of adding

One worker, several queues

A worker can have any number of queues bound to it. They all feed the same pool, so the concurrency limit covers all of them together: with concurrency 2, at most two jobs run at once, whichever queues they came from. Press Play below to watch two queues share a worker with concurrency 2, then switch to RoundRobin or bind a third queue.

Queues and a worker · simulation

busy

0 of 2

queued

6

done

0

One worker with a concurrency limit of 2 and two or three queues bound to it, each with three jobs. Press Play to watch the worker hand out the jobs one at a time: the strategy picks the queue for each free goroutine, and at most two jobs run at once, whichever queue they came from. Priority takes from the queue bound first while it has jobs. RoundRobin takes one job from each queue in turn and skips empty queues. Goroutine blocks appear when they start and disappear when they exit.

queues

pool · up to 2

done

worker := varmq.NewWorker(process, 2) emails := worker.BindQueue() reports := worker.BindQueue()

PlayStepReset

PriorityRoundRobin

Bind backups

All jobs are added before the worker takes any, and every job takes the same time. MaxLen, MinLen and queue priorities are on Strategies. A model of the library, not a benchmark.

The program below is the scene’s starting state: two queues, three jobs each, concurrency 2.

workers-and-queues/main.go

package main

import (
  "fmt"
  "time"

  "github.com/goptics/varmq"
)

func main() {
  // One worker: a pool that runs at most 2 jobs at a time.
  worker := varmq.NewWorker(func(j varmq.Job[string]) {
    fmt.Println("Processing", j.Data())
    time.Sleep(500 * time.Millisecond)
  }, 2)

  // Two queues bound to the same worker. Both feed its pool.
  emails := worker.BindQueue()
  reports := worker.BindQueue()

  for i := range 3 {
    emails.Add(fmt.Sprintf("email %d", i))
  }
  for i := range 3 {
    reports.Add(fmt.Sprintf("report %d", i))
  }

  worker.WaitUntilIdle()
  fmt.Println("All jobs done")
}

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

RunEditOpen in Go Playground

The worker has concurrency 2 and two queues: emails, bound first, and reports. Three jobs go into each. The output shows six Processing lines, at most two at a time, then All jobs done once WaitUntilIdle returns. Two jobs that start together race to print, so a pair can come out in either order, email 1 before email 0 for example.

When a slot frees up, the worker’s strategy picks which queue the next job comes from. The default strategy, Priority, takes the queue with the lowest priority number. Queues bound without WithQueuePriority tie, and a tie goes to the queue bound first, which is why every email starts before any report. WithStrategy switches to RoundRobin, MaxLen or MinLen.

With more than one queue bound, the worker hands out jobs one at a time, so the strategy decides every single job. A worker with a single in-memory queue may hand a goroutine several waiting jobs at once.

A queue belongs to one worker

Each queue is bound to the worker that created it, and Worker() on the handle returns that worker. Adding a job to a queue wakes only its own worker, and each job is processed once.

To get jobs to more than one worker, use a different source:

• Every worker gets every job: publish to a Topic and subscribe each worker with WithTopic. Topics are in-memory only.
• Workers in several processes share the jobs: bind each one to the same distributed queue with WithDistributedQueue. Each job goes to one of them.