Message Queues & Streams

The 30-second version

Sooner or later one program needs to tell another “this happened, please deal with it” — and it shouldn't have to wait, or fail, just because the other program is slow or down. That's what message queues and streams are for.

Why put something in the middle?

Imagine a sign-up page. After saving the new user it must send a welcome email, and talking to the email provider takes about 300 ms. You could call the email service directly — or publish a message “send a welcome email to ana@example.com” and let an email worker do it later. Compare the two:

That one change buys four things:

And a fourth: Fan-outFan-outDelivering one message to many recipients — one chat message pushed to all 500 people in the room.. One “order paid” message can feed the email service, the invoice service and analytics, and next month a fourth service can start listening without changing the producer at all. Inside one program this is the Observer / pub/subPublish/subscribe (pub/sub)Senders (publishers) emit messages without knowing who receives them; receivers (subscribers) sign up for the kinds of message they care about. The two sides never reference each other directly. pattern (Behavioral patterns §03); a broker does the same thing across machines.

Those are the two families of brokers: queues (RabbitMQ, Amazon SQS) and logs, often called streams (Kafka, Redis Streams). §02 puts them side by side.

What can go wrong (and what this page covers)

Putting a broker in the middle turns one function call into a little distributed systemDistributed systemA program that runs on several machines that cooperate over a network. Any machine or link can fail or slow down independently, which is what makes them hard.: several programs on several machines, any of which can crash at any moment. That raises the questions the rest of this page answers:

• Did it happen? A consumer can crash halfway through a message. Will the message be lost, or done twice? (§03 acks, §04 poison messages)
• In what order? “Paid” must not be processed before “created”. (§05)
• How do I scale it? More consumers, and how far behind they are. (§06)
• Which tool, and how? RabbitMQ routing, Kafka retention, real code. (§07–§09)
• Can I get exactly-once? What's really on offer. (§10) Then production advice, pitfalls, a quiz and a cheat sheet (§11–§14).

Two families: queues and logs

The queue: work to be done

In a queue broker like RabbitMQRabbitMQA widely used open-source message broker built around queues: messages are routed into queues and removed once a consumer acknowledges them. or Amazon SQSAmazon SQSAmazon's fully managed queue service. No servers to run; you send, receive and delete messages over HTTPS, and pay per request., a message waits in a first-in, first-outFIFOFirst in, first out: things come out in the order they went in, like a line at a shop. line. When a consumer takes it and finishes, the consumer sends an ackAcknowledgement (ack)A “got it, done” message from a receiver back to the sender (or to a message broker). If no ack arrives in time, the sender assumes the message was lost and sends it again. (“done”) and the broker deletes the message.

If several consumers read the same queue, they share the work: each message goes to just one of them. That's called competing consumersCompeting consumersSeveral consumers reading the same queue, each message going to just one of them. It's how a queue shares work between workers., and it's how you scale a queue: if one email worker can't keep up, start three.

The log: a record of what happened

In a log broker like KafkaKafkaApache Kafka: a durable, ordered log of events, split into partitions, that many programs can publish to and read from at their own pace. or Redis StreamsRedis StreamsA log data type built into Redis (5.0+): XADD appends entries, consumer groups read with XREADGROUP and acknowledge with XACK., the broker keeps an append-only logAppend-only logA list you can only add to at the end. Existing entries are never changed, so anyone can read it from any position, any number of times.: new messages are added at the end and numbered 0, 1, 2… That number is the message's offsetOffset (Kafka)A message's position number in a Kafka partition (0, 1, 2…). A consumer “commits” the offset it has finished; after a crash it restarts from the last committed offset, so anything after it is delivered again..

Reading doesn't delete anything. Each reader just remembers the offset it has reached — a bookmark. So two services can each read every message, at their own speed, and a service can move its bookmark back and read old messages again: a replayReplayReading old messages again by moving a consumer's offset back. Only logs (Kafka, Redis Streams) can do this, because they keep messages after they're read.. Messages are only deleted by retentionRetentionHow long a log keeps messages before deleting the oldest — e.g. 7 days or 500 GB per partition — whether or not anyone has read them.: “keep 7 days”, whether or not anyone read them.

Try it

Produce a few messages, then compare the two sides. On the queue side, A and B are two workers sharing the load. On the log side, A and B are two different services (say, email and analytics) reading the same log:

Things to notice:

• Queue: every message is handled by A or B, never both, and then it's gone. Replay has nothing to work with.
• Log: A and B each see every message. Their bookmarks move independently, and A can rewind to offset 0.
• A log can also share work between several copies of one service — that's a consumer groupConsumer groupA set of consumer processes that share the work of reading one stream or queue: each message (or Kafka partition) goes to exactly one member of the group., and §06 is all about it. A queue can also deliver to several services — by giving each service its own queue (§07).

Side by side

Acks, redelivery and at-least-once

Delivered is not the same as done

When a broker hands a message to a consumer, it can't know whether the consumer will finish. The consumer might crash, lose its network, or hit a bug halfway through. So the broker keeps the message, marked unacked (“in progress with consumer W1”), until it hears back:

• ack — “finished”. The broker deletes the message (queue) or the consumer moves its offset forward (log).
• nack / reject — “I failed”. The consumer chooses: put it back to be retried, or give up on it (§04).
• silence — the consumer's connection drops, or a timeout runs out. The broker assumes the worst and puts the message back for someone else. That's a redeliveryRedeliveryThe broker sending a message again because the first delivery was never acked — the consumer crashed, timed out or nacked it..

Step through the same crash twice — once with the ack after the work, once with the broker's autoAck mode, where the broker treats a message as done the moment it sends it:

1msgs, err := ch.ConsumeWithContext(ctx, "emails", "",
2    false, // autoAck off
3    false, false, false, nil)
4for d := range msgs {
5    sendEmail(d.Body) // 1. work
6    d.Ack(false)      // 2. then ack
7}

The three delivery promises

Where you put the ack picks one of these:

Why can't the broker just promise exactly-once? Look at step 4 of the stepper. The email was sent, and then the crash happened before the ack. The broker sees exactly the same thing — “no ack” — whether the consumer crashed before sending the email or after. It has to choose between maybe-lose and maybe-repeat. The safe choice is repeat.

How each broker notices a silent consumer

Kafka's version: committing offsets

Kafka has no per-message ack. (A Kafka topic is split into partitions — separate logs — and in a consumer group each partition is read by exactly one member; §05 and §06 explain both.) A consumer commits an offset — “my group has finished everything before offset 1,042 in partition 3”. The rule is the same as acking: commit after the work. Commit before it and a crash skips messages (at-most-once); commit after and a crash re-reads a few (at-least-once).

Kafka remembers only that committed number. Where the consumer had actually read up to lives in its memory and dies with it. Try both commit points:

One trap: most Kafka libraries commit for you, and some do it before your code has finished. kafkajs commits once your handler has returned for each batch — safe, if the handler awaits all its work. kafka-go's simplest read function commits each message before handing it to you. §09 shows the safe pattern in both libraries.

Prefetch: how many unacked messages at once

Sending one message, waiting for the ack, then sending the next would waste a network round trip per message. So brokers push several ahead. PrefetchPrefetch (QoS)The maximum number of unacked messages the broker will push to one consumer at a time. Prefetch 10 = “give me up to 10, then wait until I ack some”. (RabbitMQ calls it QoS) caps how many unacked messages one consumer may hold:

• Too low (1): each consumer idles for a round trip between messages. Safe and fair, but slow for quick jobs.
• Too high (unlimited — the default!): one consumer grabs thousands of messages it can't process soon, while other consumers sit idle. If it crashes, all of them are redelivered at once.
• A sensible start: about the number of messages one consumer processes in a couple of round trips — often 10–100 for fast jobs, 1–5 for slow ones. Then measure.

Prefetch is also backpressureBackpressurePushing back on whoever sends you work when you're full — making them wait or rejecting them — instead of piling it up in memory. between the broker and one consumer: a busy consumer simply isn't sent more.

Poison messages and dead-letter queues

Two kinds of failure

When a consumer fails on a message, ask why:

• Transient — the database was briefly down, a call timed out, a rate limit said “slow down”. Trying again in a few seconds will probably work.
• Permanent — the body is broken JSON, the order refers to a customer that doesn't exist, the code has a bug for this one input. Trying again will fail in exactly the same way. A message like this is a poison messagePoison messageA message that fails every time it's processed (bad data, a bug for that one input). Retrying it can never succeed and, without a limit, blocks or burns the consumer..

The consumer usually can't tell which one it is. So the standard policy covers both: retry a limited number of times, then give up and park the message somewhere safe. That “somewhere” is a dead-letter queueDead-letter queueWhere a job goes after it has failed too many times, so it stops retrying forever and a human can look at it. (DLQ).

One piece of RabbitMQ vocabulary first, because the stepper below uses it. Producers don't publish straight into a queue: they publish to an exchangeExchange (RabbitMQ)The thing RabbitMQ producers actually publish to. An exchange stores nothing; it routes each message to zero or more queues according to its type and bindings., a router that copies each message into the queues boundBindingA rule connecting an exchange to a queue: “send messages whose routing key matches X to this queue”. to it (§07 covers the routing rules). A dead-letter exchange is simply the router a queue hands its failed messages to.

1await ch.assertQueue("orders", { durable: true });
2await ch.prefetch(1); // one unacked msg at a time
3 
4await ch.consume("orders", async (msg) => {
5  try {
6    await handle(JSON.parse(msg.content.toString()));
7    ch.ack(msg);
8  } catch (err) {
9    ch.nack(msg, false, true); // requeue
10  }
11});

Compare the two modes:

• Retry forever: the poison message comes straight back and fails again, thousands of times. One consumer is stuck on it at 100% CPU; with a single consumer that may hold only one unacked message (prefetch 1, §03), every message behind it waits — head-of-line blockingHead-of-line blockingOne slow item at the front of a queue holds up everything behind it, even items that would be quick..
• Limit + DLQ: after a few attempts, the broker moves it aside. Three quick failures, then everything else flows. A person looks at the DLQ later.

The same idea in every broker

Back off between attempts

Retrying a transient failure immediately usually fails again, because whatever was broken a millisecond ago is still broken — and a thousand consumers hammering a recovering database will knock it straight back down. Wait longer after each failure: exponential backoffExponential backoffWaiting longer after each failed retry, typically doubling: 1 s, 2 s, 4 s, 8 s… up to a cap. It gives a struggling server room to recover instead of hammering it. with jitterJitterA small random amount added to a delay or timer — a retry wait, a reconnect, a connection's maximum age — so that many clients don't all act at the same instant and stampede a server. (1 s, 2 s, 4 s… plus a random bit). Async / Await §11 shows the retry loop and Worker Pools §07 covers retry budgets.

Where the wait happens matters with a broker:

• In the consumer (sleep, then retry in memory): simple, but the message is held unacked and blocks its slot — and its partition, in Kafka.
• In the broker: SQS can delay redelivery by changing the visibility timeout; RabbitMQ can dead-letter into a “wait” queue with a message TTL, which dead-letters the message back into the main queue when it expires. More moving parts, but nothing blocks.

A DLQ is a to-do list, not a bin

A DLQ nobody looks at is just a slower way to lose data. Three habits make it useful:

1. Alert on it. A gaugeGauge (metric)A metric that records a current value, which can go up and down: queue length right now, memory in use, workers busy. for DLQ depth (messages waiting in it), and an alert when it's above zero for a few minutes. §11 has the full list of metrics.
2. Keep the evidence. Store the error message, the attempt count and the original queue with the dead-lettered message (in headers), so whoever investigates doesn't have to reproduce it. RabbitMQ adds an x-death header automatically.
3. Have a redrive path. Once the bug is fixed, move the messages back to the main queue to be processed: a redriveRedriveMoving messages from a dead-letter queue back into the main queue once the problem is fixed, so they get processed for real.. SQS has a button and an API for it; with RabbitMQ and Kafka it's a small script or tool you keep ready. Redriven messages arrive late and out of order, so the consumer must tolerate that.

Ordering: per partition, per key

Why order matters, and why it's hard

Order 42 produces three events: created, paid, shipped. If the shipping service handles shipped before created, it may reject it (“unknown order”) or overwrite newer data with older data.

Within one log partition, order is easy: messages are stored and read in offset order. The trouble is that a single partition is read by a single consumer — so one partition means no parallelism at all. To get throughput, Kafka splits a topicTopic (Kafka)A named stream of messages, like “orders”. Producers write to a topic; it's split into partitions, and consumers subscribe to it. into many partitionsPartition (Kafka)One ordered slice of a topic. Messages with the same key always land in the same partition, so they stay in order relative to each other., read by many consumers at the same time. And across partitions there is no order at all.

The key decides the partition

The way out: you rarely need global order — only order per thing: per order, per customer, per bank account. So every Kafka message can carry a key, and the producer picks the partition like this:

partition = hash(key) % numberOfPartitions   // % = remainder after division

Same key → same hashHash functionA function that turns any input (a string, a file) into a fixed-size number or fingerprint. The same input always gives the same output, and small changes give very different outputs — used to spread keys across buckets or servers, and (with cryptographic hashes like SHA-256) to detect tampering. → same partition → read by the same consumer, in order. Different keys spread out over all partitions, so you still get parallelism. Try it — the widget uses Kafka's real default hash function (murmur2):

Three experiments:

1. Key = customer. Produce alice, carol, frank and oscar a few times each, then step. Every customer's messages stay in one partition and are handled in order (✓), while C1 and C2 work in parallel.
2. No key. Switch the toggle, produce alice three times, then step twice. Round-robinRound robinTaking turns in a fixed order: server 1, 2, 3, then 1, 2, 3 again. Simple and fair when every server and every request is the same size. (taking turns: p0, p1, p2, p0…) puts alice#1, #2 and #3 in three different partitions; C2 handles #3 while C1 is still busy elsewhere, and #2 comes after #3 (✗). (kafkajs and kafka-go take turns like this for unkeyed messages; the Java client sends a whole batch to one partition, then moves on. Either way, unkeyed messages are spread out.)
3. Add a partition. Back on “key = customer”, produce frank twice, press “add partition 3”, produce frank again, then step twice. % 3 became % 4, so frank's new message lands in partition 3 (owned by C2) while his older ones are still waiting in partition 1 (owned by C1) — out of order.

What breaks ordering (even with keys)

And with a queue broker, order is weaker still. A RabbitMQ queue hands messages out in order, but with two competing consumers, message 2 can finish before message 1 — and a redelivery puts message 1 back after others have already been handled. If you need strict order from a queue, you need a single consumer (RabbitMQ has a single active consumer setting that keeps a standby ready), or SQS FIFO queues with a MessageGroupId, which work just like a key: order per group, parallelism across groups.

Choosing a key

A good key has two properties at once:

• Everything that must stay in order shares it. Usually that's the id of the entity the events are about.
• There are many distinct values, spread fairly evenly. If one key carries 40% of the traffic (one giant customer, or a null key with a client that sends nulls to one partition), its partition becomes a hot partitionHot partitionOne partition (one slice of the data or of a message stream) getting far more traffic than the others, usually because one key — a huge customer, a viral chat, a null key — is very common. The one machine or consumer handling it becomes the bottleneck.: one consumer drowns while the rest are idle, and adding consumers doesn't help (§06).

Consumer groups, rebalancing and lag

How a group shares the work

In §02 each log reader had its own bookmark. A consumer group is several copies of one service sharing one set of bookmarks: one committed offset per partition. The broker splits the partitions among the members:

• 6 partitions, 2 consumers → 3 partitions each.
• 6 partitions, 4 consumers → 2, 2, 1, 1.
• 6 partitions, 8 consumers → 6 consumers get one each; 2 sit idle. A partition is never split between two members of the same group — that's what keeps per-key order (§05).

A different service (say, analytics) uses a different group name and gets its own set of offsets, so it reads every message independently. That's how one topic feeds many services.

Rebalancing: the pause nobody expects

When a member joins, leaves, or is declared dead (it stopped sending heartbeatsHeartbeat (application)A small “are you still there?” message your own protocol sends every few seconds. If a few heartbeats go unanswered, the program treats the other side as gone. for longer than the session timeoutSession timeoutHow long a consumer group waits without heartbeats from a member before declaring it dead and rebalancing (45 s by default in Kafka's Java client)., or took too long between pollsPoll (Kafka)One “give me more messages” request from a Kafka consumer to the broker. Consumers call it in a loop; taking too long between polls gets a consumer thrown out of its group. — a Kafka consumer asks the broker for messages in a loop, and each request is a poll), the group must re-share the partitions: a rebalanceRebalance (Kafka)Kafka moving partitions between the consumers of a group, e.g. when one joins, leaves or crashes. Messages whose offsets weren't committed yet are delivered again to the new owner.. In the classic eager protocol, every member stops, gives up all its partitions, and waits for the new assignment — even members whose partitions don't change.

Experiments:

1. Run at 300/s with 2 consumers. The group keeps up; lag stays near zero.
2. Switch to 600/s. Each consumer owns 3 partitions = 300/s of work but can only do 200/s, so lag climbs. Press + consumer once: during the ≈3 s rebalance nothing is processed (⏸), then 3 consumers × 200/s match the 600/s and lag stops growing — but it doesn't shrink, because there's no spare capacity to drain the backlog (including what piled up during the pause). Running at exactly 100% means never catching up. Keep adding: only partitions whose consumer has spare capacity drain, and at 6 consumers they all do.
3. Try 900/s with 5 consumers: one consumer gets two partitions and falls behind while the others are fine. 6 consumers fix it; 7 and 8 add nothing but idle consumers.
4. Try 1300/s: each partition alone gets ~217/s, more than one consumer can do. Add all 8 consumers — no help. You need more partitions or faster consumers.
5. Flip to cooperative and add a consumer: only the partitions that change owner pause.

Lag: are we keeping up?

Consumer lagConsumer lagHow far a subscriber of a message broker is behind the newest message — e.g. 5,000 messages or 3 minutes. Growing lag means the subscriber is too slow. is the gap between the newest offset in a partition and the group's committed offset: messages written but not yet processed. It's the health metric for a log consumer:

• Lag near zero, with spikes that drain — healthy. The broker absorbed a burst.
• Lag growing steadily — consumers are slower than producers. It will never recover by itself.
• Lag flat but high — you're keeping up with new messages, but everything is late by a constant amount.

Lag in messages is hard to judge (is 50,000 a lot?). Convert it to time with Little's LawLittle's LawL = λ × W: the average number of things in a system equals the rate they arrive times how long each one stays. 100 requests/s × 0.2 s each = 20 in progress at any moment. from Worker Pools §03: a new message waits roughly

wait ≈ lag ÷ processing rate      // 60,000 messages ÷ 2,000/s ≈ 30 s

Many tools report the time directly as age of the oldest unprocessed message (SQS: ApproximateAgeOfOldestMessage). Alert on that: “emails are more than 5 minutes late” means something to a human; “lag is 50,000” doesn't.

Shock absorber or hiding place?

A broker's backlog is a feature when it absorbs a burst: 5,000 orders arrive in one minute, workers clear them in the next five. It becomes a trap when work arrives faster than it's done on average. Then the backlog grows without limit, every message waits longer than the one before, and the broker hides the overload from the producers — they keep getting fast “published ✓” answers while the system falls hours behind.

Worker Pools §04–§05 shows why this happens to any queue, and why latency explodes well before 100% busy. For brokers the defences are:

• Capacity: at peak, consumers must be able to drain faster than producers write — leave headroom (≈30%) so backlogs from bursts and rebalances can drain.
• Limits: cap the queue's length or the messages' age so overload shows up as errors you can see, not as hidden delay (§07: max-length and TTL).
• Alerts: on lag in seconds and on a lag trend, not just a high threshold.

Routing, retention and limits

RabbitMQ: exchanges, bindings, routing keys

A RabbitMQ producer never writes into a queue directly. It publishes to an exchangeExchange (RabbitMQ)The thing RabbitMQ producers actually publish to. An exchange stores nothing; it routes each message to zero or more queues according to its type and bindings. with a routing keyRouting keyA short string a RabbitMQ producer attaches to each message, like “order.created.eu”. Exchanges compare it with bindings to decide which queues get the message. such as order.created.eu. The exchange stores nothing; it looks at its bindingsBindingA rule connecting an exchange to a queue: “send messages whose routing key matches X to this queue”. — rules like “order.created.* → queue email” — and drops a copy into every queue that matches. Each queue then works like §02: competing consumers, deleted on ack.

So “several services each get every message” in RabbitMQ means one queue per service, all bound to the same fanout or topic exchange. Each service's queue fills independently; a slow service only backs up its own queue.

Kafka: topics, retention and compaction

Kafka has no routing step: a producer names a topic, and the key picks the partition (§05). Filtering happens in the consumer — every group reads the whole topic and skips what it doesn't need. Instead, Kafka's interesting knob is how long messages stay:

• Delete by time or size — the default. retention.ms (7 days by default) or retention.bytes per partition. Kafka stores each partition as a series of files called segmentsSegment (Kafka)One of the files a Kafka partition is stored in. New messages go to the newest segment; retention deletes or compacts whole older segments.; old segments are deleted whether or not anyone read them. A consumer that's down for longer than the retention period comes back to find its offset gone, and starts from the oldest or newest message depending on its auto.offset.reset setting — silently skipping data if that's “latest”.
• Compact (cleanup.policy=compact) — log compactionLog compactionA Kafka cleanup mode that keeps at least the newest message for each key and deletes older ones, so the topic holds the latest state per key instead of full history. keeps at least the newest message for each key, forever. The topic stops being history and becomes a table of “current value per key” that any new service can load by reading from offset 0.

To delete a key from a compacted topic, you publish a tombstoneTombstoneA message with a key and an empty (null) value. In a compacted topic it means “delete this key”; compaction later removes the key entirely.: the key with a null value. Compaction is how Kafka stores consumer group offsets internally, and how Kafka's own tools keep their state: Kafka Connect (which copies data between Kafka and databases) and Kafka Streams (a library for processing streams).

Limits: making overload visible

A queue without limits is a memory leak that hasn't happened yet (Worker Pools §04). Queue brokers give you the same three tools as an in-process queue:

Hands-on: a real stream with Redis Streams

Kafka and RabbitMQ need a cluster to try anything. Redis StreamsRedis StreamsA log data type built into Redis (5.0+): XADD appends entries, consumer groups read with XREADGROUP and acknowledge with XACK. need only RedisRedisA popular in-memory data server: a fast key–value store and cache that also offers counters, sorted sets, pub/sub channels and streams (an append-only log for messages)., which many teams already run. It's a genuinely useful broker for moderate volumes, and it makes the concepts visible. Everything below was run against Redis 7.0; Redis 6.2 or newer has all of it.

The commands, one at a time

Start a server (redis-server), open redis-cli in another terminal, and follow along. The stream is called orders; the group of shipping workers is called shipping.

1. Append three entries. XADD orders * … adds an entry at the end. * asks Redis to choose the id, which is the current time in milliseconds plus a counter — so ids always increase:

> XADD orders * id 1 item book
"1793363659143-0"
> XADD orders * id 2 item pen
"1793363659151-0"
> XADD orders * id 3 item mug
"1793363659161-0"

2. Create a consumer group. The final 0 means “this group starts from the very first entry” ($ would mean “only entries added from now on”):

> XGROUP CREATE orders shipping 0
OK

3. Two consumers share the work. XREADGROUP GROUP shipping alice reads as consumer alice of group shipping. The > means “entries never delivered to anyone in this group”. Alice asks for 2 and gets entries 1 and 2; Bob then gets the only one left:

> XREADGROUP GROUP shipping alice COUNT 2 STREAMS orders >
1) 1) "orders"
   2) 1) 1) "1793363659143-0"
         2) 1) "id"
            2) "1"
            3) "item"
            4) "book"
      2) 1) "1793363659151-0"
         2) 1) "id"
            2) "2"
            3) "item"
            4) "pen"
> XREADGROUP GROUP shipping bob COUNT 2 STREAMS orders >
1) 1) "orders"
   2) 1) 1) "1793363659161-0"
         2) 1) "id"
            2) "3"
            3) "item"
            4) "mug"

That's competing consumersCompeting consumersSeveral consumers reading the same queue, each message going to just one of them. It's how a queue shares work between workers. inside a group (§06), with no partitions to plan: Redis hands each entry to whoever asks next. The price: entries handled by alice and bob can finish in any order (§05).

4. Ack, then look at what's still pending. Alice finishes entry 1 and Bob finishes entry 3. XPENDING shows the group's pending entries listPending entries list (PEL)Redis's per-group list of entries that were delivered to a consumer but not yet acknowledged. XPENDING shows it; XAUTOCLAIM hands stuck entries to another consumer. — delivered but not acked:

> XACK orders shipping 1793363659143-0
(integer) 1
> XACK orders shipping 1793363659161-0
(integer) 1
> XPENDING orders shipping
1) (integer) 1
2) "1793363659151-0"
3) "1793363659151-0"
4) 1) 1) "alice"
      2) "1"

Read it as: 1 entry pending; the oldest and newest pending ids are both …151-0 (entry 2); alice holds 1 of them.

5. Alice crashed. Bob takes over her stuck entry. Redis won't redeliver by itself — someone must claim it. XAUTOCLAIM orders shipping bob 1000 0 means “give bob every pending entry that's been idle for at least 1000 ms, scanning from the start”:

> XAUTOCLAIM orders shipping bob 1000 0
1) "0-0"
2) 1) 1) "1793363659151-0"
      2) 1) "id"
         2) "2"
         3) "item"
         4) "pen"
3) (empty array)
> XACK orders shipping 1793363659151-0
(integer) 1

"0-0" means the scan reached the end of the pending list. This is §03's redelivery, done by hand: if alice had sent the parcel before crashing, bob sends it again. At-least-once.

6. A second group reads everything again. Acks didn't delete anything. A new group — say analytics — starts at 0 and sees all three entries, and the stream still holds 3:

> XGROUP CREATE orders analytics 0
OK
> XREADGROUP GROUP analytics carol COUNT 10 STREAMS orders >
1) 1) "orders"
   2) 1) 1) "1793363659143-0"
   …  (entries 1, 2 and 3, same as above)
> XLEN orders
(integer) 3

That's the log side of §02: independent groups, each with its own position, and replay for free. Old entries go away only when you trim the stream — for example XADD orders MAXLEN ~ 100000 * … keeps roughly the newest 100,000 (the ~ lets Redis trim in efficient chunks). That's Redis's version of retentionRetentionHow long a log keeps messages before deleting the oldest — e.g. 7 days or 500 GB per partition — whether or not anyone has read them..

The same thing in Go and JavaScript

A producer that appends three orders, then a consumer loop that reads as one member of the group, acks after the work, and stops cleanly on Ctrl+C. Run two copies (with different consumer names) and they share the entries.

package main

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

	"github.com/redis/go-redis/v9"
)

func main() {
	// ctx is cancelled when the process gets Ctrl+C (SIGINT) or SIGTERM
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})

	// Producer: append three orders to the stream.
	for i, item := range []string{"book", "pen", "mug"} {
		id, err := rdb.XAdd(ctx, &redis.XAddArgs{
			Stream: "orders",
			Values: map[string]any{"id": i + 1, "item": item},
		}).Result()
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println("added", id)
	}

	// Create the group once. "0" = start from the oldest entry. MKSTREAM creates the stream if missing.
	err := rdb.XGroupCreateMkStream(ctx, "orders", "shipping", "0").Err()
	if err != nil && !redis.HasErrorPrefix(err, "BUSYGROUP") { // BUSYGROUP = it already exists
		log.Fatal(err)
	}

	consume(ctx, rdb, "alice")
}

func consume(ctx context.Context, rdb *redis.Client, me string) {
	for ctx.Err() == nil { // loop until Ctrl+C
		res, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    "shipping",
			Consumer: me,
			Streams:  []string{"orders", ">"}, // ">" = never delivered to this group
			Count:    10,
			Block:    2 * time.Second, // wait up to 2 s for new entries
		}).Result()
		if errors.Is(err, redis.Nil) {
			continue // timed out, nothing new
		}
		if err != nil {
			if ctx.Err() != nil {
				return // shutting down
			}
			log.Println("read:", err)
			time.Sleep(time.Second)
			continue
		}
		for _, msg := range res[0].Messages {
			fmt.Printf("%s ships order %v (%v)\n", me, msg.Values["id"], msg.Values["item"])
			// ...do the real work here...
			if err := rdb.XAck(ctx, "orders", "shipping", msg.ID).Err(); err != nil { // ← ack AFTER the work
				log.Println("ack:", err)
			}
		}
	}
}

Real output (Go with go-redis v9.18, Node 22 with the redis package v6.3; Ctrl+C pressed after the last line):

added 1793364496741-0
added 1793364496741-1
added 1793364496741-2
alice ships order 1 (book)
alice ships order 2 (pen)
alice ships order 3 (mug)
^C

Kafka and RabbitMQ clients in Go and JavaScript

Kafka

The producer sends two events for the same customer, so they share a key and stay in order (§05). The consumer is one member of the shipping group; it processes a message, then commits its offset (§03).

import "github.com/segmentio/kafka-go"

func produce(ctx context.Context) error {
	w := &kafka.Writer{
		Addr:         kafka.TCP("localhost:9092"),
		Topic:        "orders",
		Balancer:     &kafka.Murmur2Balancer{}, // ← same key → same partition (Java-compatible)
		RequiredAcks: kafka.RequireAll,         // wait until all in-sync replicas have it
	}
	defer w.Close() // flushes anything still buffered

	return w.WriteMessages(ctx,
		kafka.Message{Key: []byte("customer-42"), Value: []byte(`{"order":1,"status":"paid"}`)},
		kafka.Message{Key: []byte("customer-42"), Value: []byte(`{"order":1,"status":"shipped"}`)},
	)
}

func consume(ctx context.Context) error {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"localhost:9092"},
		GroupID: "shipping", // ← consumer group: partitions are shared among members
		Topic:   "orders",
	})
	defer r.Close() // leaves the group cleanly → partitions move at once

	work := context.WithoutCancel(ctx) // finish the message in hand even during shutdown
	for {
		m, err := r.FetchMessage(ctx) // waits for the next message; does NOT commit
		if err != nil {
			if errors.Is(err, context.Canceled) {
				return nil // shutting down
			}
			return err
		}
		if err := handle(work, m); err != nil {
			return err // no commit: after a restart we'll read it again
		}
		if err := r.CommitMessages(work, m); err != nil { // ← commit AFTER the work
			return err
		}
	}
}

What the settings mean:

• RequiredAcks: RequireAll / acks: -1. Each partition is copied to several brokers (its replicas). One of them, the partition leaderPartition leaderThe one broker that accepts writes for a Kafka partition. The partition's other replicas (followers) copy from it; if the leader dies, an in-sync follower takes over., takes the writes; the others (followers) copy from it. With acks=all the leader answers only after every in-sync replicaIn-sync replicas (ISR)The Kafka brokers that hold an up-to-date copy of a partition. With acks=all, a write is acknowledged only after every in-sync replica has it. has the message, so losing one broker doesn't lose it. acks=1 (leader only) is faster but loses messages if the leader dies before followers copy them. Pair acks=all with the topic setting min.insync.replicas=2 on a 3-replica topic: a write then fails rather than being stored on only one machine. Replication itself is explained in Replication.
• Who commits, and when. kafka-go's ReadMessage is the simpler call, but it commits each message before returning it, so a crash while handling it loses it — that's why the Go code uses FetchMessage + CommitMessages. kafkajs commits automatically once eachMessage has returned for the messages of a batch (by default after each batch); that's at-least-once as long as the handler awaits all its work. Start background work without awaiting it and kafkajs commits messages that are still in progress.
• The balancer. kafka-go's default ignores keys (§05). kafkajs's default hashes keys with murmur2, matching the Java client.
• Which Go client? kafka-go is simple and popular. franz-go (github.com/twmb/franz-go) is more complete: an idempotent producer by default, transactions, cooperative rebalancing. Confluent's confluent-kafka-go wraps the C library librdkafka.
• Which JS client? kafkajs is the most-used pure-JavaScript client, but its releases have slowed since 2023 (2.2.4 is the latest). Confluent's @confluentinc/kafka-javascript offers a KafkaJS-compatible API on top of librdkafka. Check the state of both before you choose.

RabbitMQ

One program that declares a durable queue, publishes one persistent message, and consumes with manual acks and a prefetch of 10.

import amqp "github.com/rabbitmq/amqp091-go"

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	ch, err := conn.Channel() // a lightweight session inside the TCP connection
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()

	// name, durable, autoDelete, exclusive, noWait, args
	q, err := ch.QueueDeclare("emails", true, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}

	// --- publish --- exchange "" = the default exchange: routing key = queue name
	if err := ch.PublishWithContext(ctx, "", q.Name, false, false, amqp.Publishing{
		DeliveryMode: amqp.Persistent, // write to disk, not just memory
		ContentType:  "application/json",
		Body:         []byte(`{"to":"ana@example.com"}`),
	}); err != nil {
		log.Fatal(err)
	}

	// --- consume ---
	if err := ch.Qos(10, 0, false); err != nil { // prefetch: at most 10 unacked at a time
		log.Fatal(err)
	}
	msgs, err := ch.ConsumeWithContext(ctx, q.Name, "", false /* autoAck */, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}
	for d := range msgs { // ends after Ctrl+C: the consumer is cancelled, msgs is closed
		if err := sendEmail(d.Body); err != nil {
			// Ack/Nack errors are ignored here: they only fail when the channel is
			// closing, and then the broker redelivers the message anyway.
			d.Nack(false, false) // failed: don't requeue → DLQ if one is set
			continue
		}
		d.Ack(false) // ← ack AFTER the work is done
	}
}

What the settings mean:

• Durable queue + persistent message. You need both to survive a broker restart: a durableDurable / persistentSurvives a broker restart. A durable queue's definition is kept on disk; a persistent message is written to disk instead of living only in memory. You usually want both. queue without persistent messages comes back empty. For a replicated queue that survives losing a whole broker node, declare a quorum queue (§04).
• A channelChannel (RabbitMQ)A lightweight session inside one TCP connection to RabbitMQ. Prefetch and acks are per channel. Unrelated to Go channels. is a lightweight session multiplexed over one TCP connection (part of RabbitMQ's protocol, AMQPAMQP 0-9-1The wire protocol RabbitMQ speaks: how clients open connections and channels, declare queues and exchanges, publish, consume and ack.). The library docs recommend separate connections for publishing and consuming in busy apps, so that the broker slowing a fast publisher down doesn't also stall your acks.
• The JS handler runs concurrently. amqplib doesn't wait for one async callback to finish before calling the next, so up to prefetch (10) messages are in progress at once — and can finish out of order (§05). Prefetch is your concurrency limit.

Did the broker really get it? Publisher confirms

sendToQueue and PublishWithContext return as soon as the message is written to the socket. If the broker crashes a millisecond later, the message is gone and the producer never knows. Turn on publisher confirmsPublisher confirmThe broker telling a producer “your message is safely stored” (or “I couldn't store it”). Without it, a publish can be lost and the producer never knows. and wait for them whenever losing a message matters:

if err := ch.Confirm(false); err != nil { // put the channel in confirm mode (once)
	return err
}
dc, err := ch.PublishWithDeferredConfirmWithContext(ctx, "", "emails", false, false,
	amqp.Publishing{DeliveryMode: amqp.Persistent, Body: body})
if err != nil {
	return err
}
ok, err := dc.WaitContext(ctx) // ← blocks until the broker says "stored"
if err != nil {
	return err
}
if !ok {
	return errors.New("broker nacked the message")
}

Waiting for each confirm one by one is slow (a round trip per message). Publish a batch, then wait once for all of them. Kafka producers get the same guarantee from acks — WriteMessages and send only resolve once the broker has replied.

Graceful shutdown, in one list

When the process gets SIGTERM (a deploy, a scale-down), the goal is to lose nothing and redo as little as possible. The general recipe is in Worker Pools §09; for a broker consumer it becomes:

1. Stop taking new messages — cancel the consumer (RabbitMQ) or stop fetching (Kafka).
2. Finish what's in hand, and ack or commit each message as it completes.
3. Close the consumer properly. Kafka: the member leaves the group, so its partitions are reassigned immediately instead of after the session timeout (§06). RabbitMQ: closing the channel requeues anything still unacked — nothing is lost, it's just redelivered.
4. Give up after a deadline (your platform kills you anyway — Kubernetes after 30 s by default). Anything unfinished will be redelivered: one more reason handlers must be idempotent.

Exactly-once: what's really on offer

Where duplicates (and losses) come from

Follow one “order paid” event from the database to the email it triggers. Things can go wrong in three places:

1. The producer's two writes. The order service saves to its database and publishes to the broker. A crash between the two loses the event or announces something that didn't happen.
2. The producer's retries. The broker stored the message, but the ack got lost; the producer retries and the message is stored twice.
3. The consumer's crash. The consumer did the work but crashed before acking (§03) — or the message was redriven from a DLQ, or replayed. It runs again.

Each has its own fix. Only the last one is fully in your hands, and it covers for the other two.

1. Database + broker: the outbox

You can't make a database write and a broker publish one atomic step. The standard answer is the transactional outboxOutbox patternWrite an outgoing event into an outbox table in the same database transaction as the change itself; a separate process reads that table and publishes the events. Either both happen or neither does.: in the same database transactionDatabase transactionA group of database changes that happen all together or not at all (COMMIT keeps them, ROLLBACK undoes them). How much other users can see or interfere with the half-done state depends on the isolation level. as the business change, insert the event into an outbox table; a separate relay process (or change data captureChange data capture (CDC)Reading a database's own change log (in Postgres, the write-ahead log) as a stream of every insert, update and delete, so another system can copy the changes without the app doing anything. reading the database log) publishes outbox rows to the broker and marks them sent. Behavioral patterns §10 has a crash simulator for exactly this, and Sagas builds multi-step processes on it.

Note what the outbox guarantees: the event is published at least once. The relay can crash after publishing but before marking the row sent, and publish it again. Duplicates again — handled by fix 3.

2. Producer retries: the idempotent producer

Kafka's idempotent producerIdempotent producerA Kafka producer mode where the broker removes duplicates caused by the producer's own retries, using a producer id plus a sequence number on every batch. numbers each batch per partition, so the broker can drop a retried batch it has already written. It also keeps retries from reordering messages. It's on by default in the Java client since Kafka 3.0 and in franz-go; kafkajs needs idempotent: true; kafka-go doesn't support it. RabbitMQ and SQS have no equivalent (SQS FIFO queues drop repeats with the same MessageDeduplicationId within a 5-minute window).

It only covers the client's own retries within one producer session. A restarted producer, an outbox relay that publishes a row twice, or your code calling send() twice all produce real, distinct duplicates.

3. Kafka → Kafka: transactions

A common Kafka job reads from one topic, computes something, and writes to another: orders in, invoices out. Here a crash can cause a duplicate output: the invoice was written, but the input offset wasn't committed, so the input is processed again.

Kafka transactionsKafka transactionsLets a producer write to several partitions — and commit consumer offsets — as one all-or-nothing unit. Readers with isolation.level=read_committed never see aborted writes. close that gap by putting the output messages and the input offset commit in one atomic unit:

const producer = kafka.producer({
  transactionalId: "invoicer-1", // stable per instance: if an old copy of this worker is
                                  // still running (it froze and was replaced), Kafka rejects its writes
  idempotent: true,
  maxInFlightRequests: 1,
});

await consumer.run({
  autoCommit: false, // the transaction commits the offset instead
  eachMessage: async ({ topic, partition, message }) => {
    const tx = await producer.transaction();
    try {
      await tx.send({ topic: "invoices", messages: [{ key: message.key, value: makeInvoice(message.value) }] });
      await tx.sendOffsets({          // "I've consumed up to here" — inside the same transaction
        consumerGroupId: "invoicer",
        topics: [{ topic, partitions: [{ partition, offset: (BigInt(message.offset) + 1n).toString() }] }],
      });
      await tx.commit();              // output + offset become visible together, or not at all
    } catch (err) {
      await tx.abort();
      throw err;
    }
  },
});

Downstream consumers must read with isolation.level=read_committed (kafkajs: readUncommitted: false, the default), or they'll see writes from aborted transactions. Kafka Streams and Flink use this machinery when you turn on “exactly-once”.

The honest fine print: this is exactly-once only for state that lives in Kafka. If makeInvoice also sends an email, charges a card or writes to Postgres, those effects are not rolled back when the transaction aborts and retries. Transactions also add latency (each commit is extra round trips), so they're usually committed per batch, not per message as in this short example.

4. The fix that always works: an idempotent consumer

Whatever happened upstream, the consumer can make its effect happen once. The simplest robust version: keep a table of message ids you've processed, and insert the id in the same database transaction as the effect. If the insert finds the id already there, it's a duplicate — skip the work, but still ack.

// handleOnce runs apply at most once per message id: the id and the effect commit together.
func handleOnce(ctx context.Context, db *sql.DB, msgID string, apply func(*sql.Tx) error) (string, error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return "", err
	}
	defer tx.Rollback() // undoes everything unless Commit ran first

	res, err := tx.ExecContext(ctx,
		`INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, msgID)
	if err != nil {
		return "", err
	}
	if n, err := res.RowsAffected(); err != nil {
		return "", err
	} else if n == 0 { // ← id already there: a duplicate
		return "duplicate, skipped", nil
	}
	if err := apply(tx); err != nil { // the real change, in the SAME transaction
		return "", err
	}
	return "applied", tx.Commit()
}

We ran both against PostgreSQL 16, delivering the same message twice to an apply that adds 10 loyalty points (Go with pgx/v5, Node with pg):

applied
duplicate, skipped
ana's points: 10

What makes this work, and its limits:

• Which id? It must be the same on every redelivery, and different for different messages. Kafka's topic-partition-offset (e.g. orders-3-1042 for offset 1,042 in partition 3 of orders) works for one consumer group. A business id chosen by the producer (an idempotency keyIdempotency keyA unique ID the client sends with a request (e.g. an Idempotency-Key header) so the server can recognise a repeat and do the work only once. or event id stored in the message) is better: it also catches duplicates the outbox relay or a producer retry created, which have new offsets.
• Same transaction, same database. If the effect is in another system — an email, a payment API — you can't commit it together with the id. Then pass the key along (most payment providers accept an idempotency key) or record “started” and “done” states. Idempotency covers these cases.
• Clean up old ids eventually (e.g. after 30 days — longer than any possible redelivery or replay window).

Choosing and running a broker

Which broker?

The throughput figures are orders of magnitude only; they depend heavily on message size, durability settings and hardware. Most teams hit operational limits — people who know how to run and debug the thing — long before raw throughput limits.

Messages are a public API: schema evolution

The producer and its consumers are deployed separately, by different teams, at different times. With a log, a consumer may even read a message written months ago. So the message format is a contract you change carefully — the same problem gRPC solves for requests (gRPC §05).

The rules that keep old and new code compatible:

• Add fields; never remove or rename one that's in use. New fields get defaults, so old messages still parse. Old consumers ignore fields they don't know (check that your JSON decoder does).
• Never change a field's meaning or type. amount in cents must not become amount in euros. Add amount_eur instead.
• Deploy consumers before producers when a new field matters, so nobody receives something they can't handle.
• Put a type and version in every message ("type": "order.paid", "version": 2) so a consumer can route or reject what it doesn't understand instead of crashing on it — which would make it a poison message (§04).

Binary formats with schemas — ProtobufProtocol Buffers (protobuf)Google's compact binary data format: you describe messages in a .proto file and a tool generates Go/JS code to encode and decode them. Smaller and faster than JSON; used by gRPC., AvroAvroA compact binary data format with a schema, like Protobuf. Common with Kafka and schema registries. (a similar binary format, common with Kafka) — turn these rules into tooling. A schema registrySchema registryA small service that stores the versions of each message schema (Avro, Protobuf, JSON Schema) and rejects changes that would break existing consumers. stores every version of each topic's schemaSchemaA description of what shape data must have: which fields exist, their names and types. A database table has one, a .proto file is one for gRPC messages, and message brokers often keep schemas in a registry so producers and consumers agree. and rejects a producer change that would break compatibility before it ships.

What to monitor

The hidden costs of event-driven systems

Event-drivenEvent-driven architectureServices that react to events published by others (“order paid”) instead of calling each other directly. Loosely coupled, but the flow is harder to see and debug. systems are loosely coupled in the code. That's the point — and it hides some costs:

• Coupling doesn't disappear; it moves into the message format. Ten services parse order.paid, so changing it is a ten-team migration. Decide who owns each topic and who may publish to it.
• The flow becomes invisible. “What happens when an order is paid?” is no longer a function you can read; it's whoever subscribes today. Keep a catalogue of topics, producers and consumers.
• Debugging needs tracing. Put a correlation id (or trace context) in every message's headers and log it everywhere, so you can follow one order across five services and two brokers. Distributed tracing makes this visual — see Observability.
• Eventual consistency becomes user-visible. Right after “Pay”, the order page may still say “pending” because the update travels through a queue. Design the UI for it; Replication §06 shows the same effect with database replicas.
• Not everything should be async. If the caller needs the answer to continue (“is this card valid?”), a direct call with a timeout is simpler, and fails more honestly, than a request/response dance over a broker.

Pitfalls gallery

eachMessage: async ({ message }) => {
  processInBackground(message); // ← no await: returns at once
}

db.Exec(`UPDATE orders SET status = 'paid' WHERE id = $1`, id)
w.WriteMessages(ctx, orderPaid(id)) // crash between these two lines = nobody ever hears

Check yourself

Predict the answer before you click. Q1 is covered in §02 and §07, Q2 in §03, Q3 in §06, Q4 in §05, Q5 in §06, and Q6 in §10.

for d := range msgs { // autoAck = false
    d.Ack(false)
    if err := charge(d.Body); err != nil {
        log.Println(err)
    }
}

Cheat sheet