Job Queues
A channel is a queue that dies with the process; a durable queue is rows that survive a restart.
SELECT id, payload FROM jobs WHERE state = 'ready' ORDER BY run_at FOR UPDATE SKIP LOCKED LIMIT 1A job queue is a list of work waiting to be done. Workers take items off it and do them. The interesting question is where that list lives.
A channel is a queue in memory
A buffered channel is the simplest queue there is. It costs nothing to set up and it is very fast. It also lives inside one process. If that process is killed, restarted or redeployed, everything still in the channel is gone, and nobody is told. For work you can afford to lose, such as warming a cache, that is a fair trade.
Durable queues
Durable means the job is written to storage before the queue accepts it: a row in Postgres, an entry in a Redis list, or a message in a managed service. A restart loses nothing, because the job is data. Many processes can share one queue, so you add capacity by adding replicas.
A worker does not just read a job, it claims it: one statement marks the row as taken so no other worker picks it up. In Postgres 9.5 and later, and MySQL 8.0 and later, that is SELECT ... FOR UPDATE SKIP LOCKED. The claim lasts for a limited time. If the worker dies, the claim expires and the job becomes available again.
What trips people up
That expiry is exactly why a durable queue delivers at-least-once: a job that was finished but never marked finished will run a second time. Every job must be safe to run twice. Store an attempt count and a run_at time too, so a failing job backs off instead of spinning.
Example
package main
import (
"fmt"
"sync"
)
// memQueue is a queue that lives in memory: fast, and lost on restart.
type memQueue struct{ jobs chan string }
// enqueue reports whether the job was accepted.
func (q *memQueue) enqueue(payload string) bool {
select {
case q.jobs <- payload:
return true
default:
return false // full, and there is nowhere else to put it
}
}
func main() {
q := &memQueue{jobs: make(chan string, 2)} // room for two waiting jobs
for _, p := range []string{"email:1", "email:2", "email:3"} {
fmt.Printf("enqueue %s accepted=%t\n", p, q.enqueue(p))
}
var wg sync.WaitGroup
wg.Go(func() {
for p := range q.jobs {
fmt.Println("handled", p)
}
})
close(q.jobs) // the worker drains what is buffered, then stops
wg.Wait()
}
// Output:
// enqueue email:1 accepted=true
// enqueue email:2 accepted=true
// enqueue email:3 accepted=false
// handled email:1
// handled email:2
When to use it
- Finish a payment webhook's follow-up work even though the pod restarted mid-job.
- Add throughput by scaling replicas, with each job going to exactly one worker.
- Retry a failing export after a delay instead of hammering it in a loop.
- Park a job that has failed five times so somebody can read the error.
More examples
The same worker loop, shorthand and longhand
Most Go code uses the range form, so you need to recognise it, but the longhand is what it does. Write the longhand when you need more than the payload in the loop, for example a select that also watches a shutdown channel.
jobs := make(chan string, 64) // a 64-slot in-memory queue
// Shorthand: ranging over a channel ends when the channel is closed and drained.
go func() {
for payload := range jobs {
handle(payload)
}
}()
// Longhand: the same loop written out. The second value from a receive is
// false once the channel is closed and empty. These two are the same thing.
go func() {
for {
payload, ok := <-jobs
if !ok {
return
}
handle(payload)
}
}()Claiming one job from a Postgres queue
This is a complete durable queue in one function: the database is the queue, and the transaction is what stops two workers running the same job. Reach for it when you already have Postgres and do not want another piece of infrastructure to operate.
// claim takes a single job and marks it running, inside one transaction.
// SKIP LOCKED makes two workers pick different rows instead of waiting on
// each other (Postgres 9.5+, MySQL 8.0+).
func claim(ctx context.Context, db *sql.DB) (int64, string, error) {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return 0, "", err
}
defer tx.Rollback() // after a successful Commit this returns sql.ErrTxDone and changes nothing
var id int64
var payload string
err = tx.QueryRowContext(ctx, `SELECT id, payload FROM jobs
WHERE state = 'ready' AND run_at <= now()
ORDER BY run_at
FOR UPDATE SKIP LOCKED
LIMIT 1`).Scan(&id, &payload)
if err != nil {
return 0, "", err // sql.ErrNoRows simply means the queue is empty
}
if _, err := tx.ExecContext(ctx,
`UPDATE jobs SET state = 'running' WHERE id = $1`, id); err != nil {
return 0, "", err
}
return id, payload, tx.Commit()
}Backing off, then giving up
Each retry waits twice as long as the last, and the fifth failure moves the job to a dead state where a person can look at it. Without this, one permanently broken job occupies a worker in a tight loop and fills the log.
// fail returns the job to the queue with a delay, or parks it after 5 attempts.
func fail(ctx context.Context, db *sql.DB, id int64, cause error) error {
_, err := db.ExecContext(ctx, `UPDATE jobs SET
attempts = attempts + 1,
last_error = $2,
state = CASE WHEN attempts + 1 >= 5 THEN 'dead' ELSE 'ready' END,
run_at = now() + (interval '10 seconds' * power(2, attempts))
WHERE id = $1`, id, cause.Error())
return err
}
Discussion