Workers and queues

A worker runs jobs; a queue holds them until the worker has a free slot. To give a worker a queue, you ask the worker for one, 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):

Which queue to bind

• FIFO in-memory queue (BindQueue): jobs leave the queue in the order you add them. The queue lives only while the process runs. See Queues.
• Priority queue (BindPriorityQueue): each job has an int priority, and a lower number runs first. See Queues.
• Persistent queue (WithPersistentQueue): one process wants its jobs to survive a restart. See Persistence and distribution.
• Distributed queue (WithDistributedQueue): workers in several processes share one queue, and each job goes to one of them. See Persistence and distribution.
• Topic (WithTopic): every subscribed worker must get its own job for each payload. Topics are in-memory only. See Topics.

A process that only adds jobs does not need a worker. NewDistributedQueue wraps a distributed queue’s storage with no worker bound to it.

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 the same time, from any of the queues. Press Play below to watch two queues share a worker with concurrency 2, then switch to RoundRobin or bind a third queue.

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

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

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 together.

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.

Next: jobs.