Queues, Topics and Streams
How a broker passes messages between programs, and the difference between one-consumer queues, pub/sub topics and retained streams.
queue: one message -> one consumer; topic: one message -> every subscriberA message broker is a server that sits between programs and holds messages for them. One program hands it a message and carries on working. Another program picks the message up later. The two never have to be running at the same moment. The program that sends is the producer. The program that reads is the consumer.
Queues
A queue is point-to-point. Each message is delivered to exactly one consumer. Start a second consumer and the work is shared, not duplicated. This is what you want for jobs: send one welcome email, charge one card, resize one image.
Topics
A topic is publish-subscribe, usually written pub/sub. One message is copied to every subscriber. The sender does not know who is listening, and adding a listener needs no change to the sender. Use a topic when one fact has several independent readers: billing, search indexing and analytics all care that an order was paid.
Streams
A stream is a topic that keeps its messages after they have been read. The broker stores them in order as a log that only grows at the end. Each message has an offset, a number giving its position. A consumer remembers its own offset, so it can stop, restart and carry on. A brand new consumer can start at offset zero and replay the whole history.
What trips people up
The three words are not used consistently. Kafka calls everything a topic, but a Kafka topic is a stream, and consumer groups give it queue behaviour on top. NATS core subjects are topics with no storage at all. Read the broker's own definition before you rely on the word.
Example
package main
import (
"fmt"
"sort"
"sync"
)
func main() {
// Queue behaviour: three messages, two workers, each message handled once.
jobs := make(chan string, 3)
for _, m := range []string{"a", "b", "c"} {
jobs <- m
}
close(jobs)
var mu sync.Mutex
var handled []string
var wg sync.WaitGroup
// "for range 2" is the Go 1.22 short form of "for i := 0; i < 2; i++".
for range 2 {
wg.Add(1)
go func() {
defer wg.Done()
// Whichever worker is free takes the next message.
for m := range jobs {
mu.Lock()
handled = append(handled, m)
mu.Unlock()
}
}()
}
wg.Wait()
sort.Strings(handled) // workers finish in any order, so sort before printing
fmt.Println("queue, each handled once:", handled)
// Topic behaviour: one message, a copy for every subscriber.
for _, name := range []string{"email", "search"} {
fmt.Println("topic, copy for", name+":", "order.paid")
}
}
// Output:
// queue, each handled once: [a b c]
// topic, copy for email: order.paid
// topic, copy for search: order.paidWhen to use it
- Send a welcome email after signup without making the signup request wait for the mail server.
- Tell billing, search and analytics that an order was paid, from one publish.
- Let a new reporting service rebuild its tables by replaying every order event from the start.
- Keep an image-resizing worker busy while the web server returns immediately.
More examples
A stream: one log, one offset per consumer
This is the whole idea of a stream in twenty lines: the messages stay put, and each consumer owns its own position. It explains two things you will meet on a real broker. A consumer that restarts resumes where it stopped, and a new consumer can be pointed at offset zero to rebuild its view from history. Reach for a stream whenever replay has value, such as filling a new search index from events that were already published.
package main
import "fmt"
// A stream is an append-only log plus one remembered position per consumer.
type stream struct {
log []string
offsets map[string]int // consumer name -> next index to read
}
func (s *stream) read(consumer string) []string {
// A missing map key gives 0, so a new consumer starts at the beginning.
start := s.offsets[consumer]
s.offsets[consumer] = len(s.log)
return s.log[start:]
}
func main() {
s := &stream{
log: []string{"created", "paid", "shipped"},
offsets: map[string]int{},
}
// billing reads everything stored so far.
fmt.Println("billing:", s.read("billing"))
s.log = append(s.log, "delivered")
// billing resumes after "shipped" instead of re-reading.
fmt.Println("billing:", s.read("billing"))
// search has never read before, so it replays the whole history.
fmt.Println("search:", s.read("search"))
}
// Output:
// billing: [created paid shipped]
// billing: [delivered]
// search: [created paid shipped delivered]Fan-out: a separate copy per subscriber
A topic is a copy per subscriber, and that is what makes a slow subscriber dangerous: the copies pile up somewhere. Here the buffer of 8 is that somewhere, and a real broker has the same problem with a bigger buffer. Use this shape in a single process when several unrelated parts of your service need to react to one event, before you reach for a broker at all.
package main
import (
"fmt"
"sort"
)
// broker holds one channel per subscriber.
type broker struct {
subs map[string]chan string
}
func (b *broker) subscribe(name string) {
// Buffered, so one slow subscriber does not block the publisher.
b.subs[name] = make(chan string, 8)
}
func (b *broker) publish(msg string) {
for _, ch := range b.subs {
ch <- msg // every subscriber gets its own copy
}
}
func main() {
b := &broker{subs: map[string]chan string{}}
b.subscribe("email")
b.subscribe("search")
b.publish("order.paid")
names := make([]string, 0, len(b.subs))
for name := range b.subs {
names = append(names, name)
}
sort.Strings(names) // map iteration order is random, so sort it
for _, name := range names {
fmt.Println(name, "received", <-b.subs[name])
}
}
// Output:
// email received order.paid
// search received order.paidRedelivery when a handler fails
A queue is not just a list; it is a list where a message that was not finished comes back. That single behaviour is why brokers hand you duplicates and why handlers have to tolerate running twice. Keep this picture in mind when you set a retry limit: without one, a message that always fails is retried forever and blocks everything behind it.
package main
import "fmt"
func main() {
queue := []string{"job-1", "job-2"}
attempts := map[string]int{}
for len(queue) > 0 {
job := queue[0]
queue = queue[1:]
attempts[job]++
// job-2 fails the first time and goes back on the queue.
if job == "job-2" && attempts[job] == 1 {
fmt.Println("failed, requeued:", job)
queue = append(queue, job)
continue
}
fmt.Println("done:", job, "attempts:", attempts[job])
}
}
// Output:
// done: job-1 attempts: 1
// failed, requeued: job-2
// done: job-2 attempts: 2
Discussion