Redis

redisq is the Redis adapter for VarMQ. It provides first-in first-out and priority queues on one Redis server, each in a persistent and a distributed form. Read persistence and distribution first for what the two kinds of queue do.

Setup

redisq needs Redis 6.0 or newer, with Lua scripts allowed: it takes, adds and puts back jobs with scripts. Its go.mod asks for Go 1.26. Three steps set it up:

1. 1

   Start Redis

   varmq’s repository has examples/compose.yml, which starts Redis 7.2 on port 6375. The programs on this page use that port.

   From a cloneWithout a clone

   terminalCopy

   docker compose -f examples/compose.yml up -d

   Run it from a clone of github.com/goptics/varmq.

2. 2

   Add the two modules

   varmq’s version is the release these pages describe.

   terminalCopy

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

3. 3

   Create a queue and bind it

   redisq.New takes a Redis URL and returns a Queues value. It panics when the URL does not parse. Each queue takes a key, which names it in Redis. Pick the kind of queue.

   PersistentPersistent priorityDistributedDistributed priority

   main.goCopy

   rdb := redisq.New("redis://localhost:6375")
   defer rdb.Close()
   
   pq := rdb.NewQueue("persistent_queue")
   defer pq.Close()
   
   queue := worker.WithPersistentQueue(pq)

   First in, first out, on a Redis list. One process keeps its jobs across a restart.

Queues has one constructor for each kind of queue. Each takes a key, which names the queue in Redis:

A queue with the key jobs keeps its waiting jobs under jobs. Every other key it uses, and the wake-up channels of a distributed queue, start with jobs:. Give each queue its own key.

Redis keeps these in memory. Redis’s own persistence settings decide whether they survive a restart of Redis itself. Neither VarMQ nor redisq changes these settings.

A persistent queue

The program below binds a persistent queue to a worker with concurrency 1, adds ten jobs, and waits for them.

11 collapsed lines
package main

import (
	"fmt"
	"time"

	"github.com/goptics/redisq"
	"github.com/goptics/varmq"
)

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

	pq := rdb.NewQueue("persistent_queue")
	defer pq.Close()

	worker := varmq.NewWorker(
		func(j varmq.Job[int]) {
			fmt.Println("processing", j.Data())
			time.Sleep(time.Second)
			fmt.Println("processed", j.Data())
		},
		varmq.WithAutoRun(false),
	)

	queue := worker.WithPersistentQueue(pq)
	fmt.Println("left in Redis by an earlier run:", queue.Len())

	// Start after the bind, so the worker looks at jobs already in Redis.
	worker.Start()

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

	worker.WaitAndStop()
}

Three choices in it matter:

• The worker is created with WithAutoRun(false) and started after the bind. Binding a redisq queue already makes the worker look for jobs that a previous run left in Redis. The overview says this. Starting the worker later only delays it while the program counts the jobs.
• queue.Len() is read before Start. It asks Redis, so it counts the waiting jobs from the earlier run.
• Every job finishes before the deferred closes run. WaitAndStop lets every job run, then stops the worker. A job acknowledges when its function returns. Close on a redisq queue puts back the jobs it took and has not acknowledged. If you close it while a job runs, that job goes back to the queue to run again. The job’s acknowledgement then fails.

Run locally

1. Start Redis, as in Setup.

2. Make a module, save the program as main.go and add the dependencies:

   mkdir redis-persistent && cd redis-persistent
   go mod init redis-persistent
   go get github.com/goptics/varmq github.com/goptics/redisq

   Copy

3. Run it with go run .. The first run prints left in Redis by an earlier run: 0, then a processing line and a processed line for each job, one second apart. Concurrency is 1, so they run in the order they were added.

4. Run it again, and press Ctrl+C a few seconds in. The jobs that had not started are still in Redis.

5. Run it a third time. It prints how many jobs are left, and the worker runs those first, before the ten new ones.

The job that was running when you pressed Ctrl+C is not in that count. It waits in persistent_queue:nacked until redisq treats the run that took it as dead. This happens once the program has run again for 30 seconds in total. Run the program once or twice more, and the job comes back at the front of the queue. The section on unacknowledged jobs explains the timing.

A priority queue

NewPriorityQueue gives a queue that gives out the lowest number first. WithPersistentPriorityQueue binds it, and Add on the handle takes the priority after the data.

10 collapsed lines
package main

import (
	"fmt"

	"github.com/goptics/redisq"
	"github.com/goptics/varmq"
)

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

	pq := rdb.NewPriorityQueue("persistent_priority_queue")
	defer pq.Close()

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

	queue := worker.WithPersistentPriorityQueue(pq)

	// A lower number runs first.
	queue.Add("report", 10)
	queue.Add("alert", 1)
	queue.Add("newsletter", 5)

	worker.Start()
	worker.WaitAndStop()
}

Run locally

Follow the steps for the persistent queue, with this program as main.go. It prints:

processing alert
processing newsletter
processing report

redisq keeps a priority queue in a Redis sorted set, with the priority as the score. It stores each job with a sequence number in front. So identical jobs stay separate, and jobs with the same priority come out in the order they were added. A job that goes back to the queue keeps its priority.

A distributed queue

A distributed queue is shared by any number of processes. The producer below has no worker. It wraps the Redis queue with varmq.NewDistributedQueue and adds jobs.

10 collapsed lines
package main

import (
	"fmt"

	"github.com/goptics/redisq"
	"github.com/goptics/varmq"
)

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

	dq := rdb.NewDistributedQueue("distributed_queue")
	defer dq.Close()

	// No worker here: this process only adds jobs.
	producer := varmq.NewDistributedQueue[string](dq)

	for i := range 20 {
		if !producer.Add(fmt.Sprintf("https://example.com/%d", i)) {
			fmt.Println("not added:", i)
		}
	}

	fmt.Println("waiting in Redis:", producer.Len())
}

The consumer binds the same Redis queue to a worker with concurrency 5, and runs until you stop it.

15 collapsed lines
package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/goptics/redisq"
	"github.com/goptics/varmq"
)

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

	dq := rdb.NewDistributedQueue("distributed_queue")
	defer dq.Close()

	worker := varmq.NewWorker(
		func(j varmq.Job[string]) {
			fmt.Println("scraping", j.Data())
			time.Sleep(2 * time.Second)
			fmt.Println("scraped", j.Data())
		},
		5,
		varmq.WithAutoRun(false),
	)

	queue := worker.WithDistributedQueue(dq)
	fmt.Println("waiting in Redis:", queue.Len())

	worker.Start()
	fmt.Println("listening")

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	<-ctx.Done()

	// Let running jobs finish and acknowledge before the queue closes.
	worker.StopAndWait()
}

How it fits together:

• A job reaches one consumer. The consumer takes a job from the Redis list. After that, the job is not in the list for another consumer. Two consumers on one key divide the jobs between them.
• Add wakes an idle consumer. Enqueue on a distributed queue adds the job. In the same step, it wakes consumers that are waiting for work, at most one per waiting job. Each consumer has a channel of its own. A consumer waits for work once it finds the queue empty. A busy consumer is not woken, so the wake-up goes to a consumer that can take the job. While jobs wait, every consumer also checks once a second. So a consumer still takes a job that was added while no consumer was listening.
• The producer has a Len and a Purge. On a distributed handle they are redisq’s own. Len is the length of the Redis list. Purge deletes the list, which removes every waiting job. A job that a consumer took and is running is not in the list, and Purge does not touch it.
• Each consumer’s counters are its own. Jobs run in one process and are added in another. So one instance’s Submitted and Completed need not match. The overview says what is known.
• The consumer stops its worker before the deferred Close. Close puts back every job the consumer took and has not acknowledged. It also wakes another consumer for them. So if you close while a job runs, the job runs twice. StopAndWait lets the running jobs finish and acknowledge first. redisq has a Listen method that blocks until SIGINT or SIGTERM. It does not stop the worker.

Run locally

1. Start Redis, as in Setup.

2. In one module, save the producer as producer/main.go and the consumer as consumer/main.go:

   mkdir redis-distributed && cd redis-distributed
   go mod init redis-distributed
   mkdir producer consumer
   go get github.com/goptics/varmq github.com/goptics/redisq

   Copy

3. In one terminal run go run ./consumer. It prints waiting in Redis: 0 and listening.

4. For a second consumer, run go run ./consumer in another terminal. Each job goes to one of the two.

5. In a third terminal run go run ./producer. It prints waiting in Redis: and a number. The number is 20 when no consumer is running, and lower when one has already taken some jobs.

6. The consumers print a scraping line and, two seconds later, a scraped line for each job. Each consumer runs up to five jobs at a time, so five lines appear together.

7. Press Ctrl+C in a consumer terminal. It finishes the jobs it is running, acknowledges them and exits.

To see a crash instead, run the producer again and kill one consumer with kill -9 while it runs jobs. The other consumer runs them a few seconds later, as the next section explains.

Jobs that were not acknowledged

The overview describes when VarMQ acknowledges a job. This is what redisq does with a job that was taken and not acknowledged, for example because the process died.

When such a job runs again:

• Close gives jobs back. Close puts back at once every job its queue object took and has not acknowledged. After it, Enqueue and Acknowledge on that object return false. See shutting down.
• The timeouts are set per queue object. SetVisibilityTimeout changes the deadline for jobs taken after the call. SetHeartbeatTimeout sets how long the object may go without reaching Redis: at least five seconds, and 0 turns heartbeats off. Call them right after you create the queue.
• Keep the visibility timeout longer than your slowest job. A job still running past its deadline is put back by the next sweep and runs again.
• Keep the heartbeat timeout longer than any freeze you want to survive. A process can be frozen or cut off from Redis for longer than this timeout. Then redisq treats it as dead. Its running jobs run again elsewhere, and their later Acknowledge still returns true.
• GetNackedItemsCount counts the jobs taken and not acknowledged. The handle’s Len does not, because they have left the queue.

How redisq tracks a taken job

The scene below shows the keys of a queue named jobs. Take a job, then kill the process or close the queue, and move the slider to move time forward.

• A taken job is recorded in the same step. One script moves the job from the queue to the hash <key>:nacked, under its acknowledgement ID. The same script gives the job a deadline in <key>:timeouts. The deadline is Redis’s time plus the visibility timeout, five minutes by default. Acknowledge removes both entries.
• Every queue object sweeps once a second. From its creation until Close, it puts back the jobs whose deadline has passed. It also puts back the jobs held by a queue object that is treated as dead. On the first-in first-out queues they go to the front, and on a priority queue back at their priority. On a distributed queue the sweep wakes a consumer for them. Creating a queue object also puts back the jobs whose deadline has passed.
• A queue object is treated as dead when its heartbeat runs out. Each queue object renews a heartbeat in <key>:heartbeats when it takes jobs and on every sweep. The other queue objects on the key put back its jobs once the heartbeat is 30 seconds old, the default. Time in which no process swept the key does not count, because Redis cannot tell a dead process from a pause of all of them. A distributed consumer is treated as dead sooner. Redis drops its wake-up channel when its process exits. Five seconds later, the other consumers put its jobs back.

Job IDs

redisq checks job IDs in EnqueueOnce. An Add with WithJobId stores the job once per ID within a window, five minutes by default. Another Add with the same ID in that time stores nothing, and still returns true. That covers an Add you retry, and a push that the Redis client sent again after its reply was lost. The ID is remembered for the whole window, even after its job has run. SetDedupWindow on the queue object changes the window, and 0 turns the check off.

A job without an ID is not checked. If the reply to its push is lost, the Redis client sends the push again, and the job can be stored twice. Give jobs IDs when that matters.

Batches

redisq implements IBatchAcknowledgeable. So a worker with only this queue takes as many jobs as it has free slots in one call. Each job starts on its own goroutine at once.