--- title: 'Message Queues and Streams' source: 'https://academia.sh/en/courses/architectural-styles/message-queues-and-streams' course: 'Architectural Styles' language: en updated: '2026-08-23T07:01:09+00:00' license: 'CC BY-SA 4.0' --- # Message Queues and Streams Measuring the effect of putting a store between the publisher and the consumer: the consumer work unit run the moment a message is dropped, the number of pending messages when the consumer falls behind the producer, whether order is preserved with one and two consumers, the delivery record opened when a message is delivered a second time, and the number of history records a unit added afterward can read from a queue versus a stream. The previous lesson's bus had an unmeasured property: the `publish` call did not return until every attached handler finished. Four handlers ran in sequence on the same call stack, so the module that finalizes the fee took as long as the slowest handler. If archive writing slowed down, fee finalization slowed down too, even though the archive's write time does not matter for pricing. This lesson puts a **store** in between: the publisher drops the message and returns, the consumer takes it later. The **message queue** and the **stream** were built and measured in the Messaging topic of the Caching, Queues and Asynchronous Processing course: delivery semantics, the ordering/partitioning trade-off, dead-letter queues, and retry belong there and are not revisited here. What falls to this lesson is the style itself — what the store guarantees, what it does not, and the constraint this brings to the architecture. Four things are measured: consumer work run at the drop, pending messages, whether order is preserved, and the consequence of the same message processed twice. The **quality attribute** it connects to is reliability, and the scenario question is: when the consumer falls behind the producer, or a message is delivered a second time, how much work piles up and how many delivery records get opened twice. A second measure falls into the performance efficiency family: how many consumer work units have run by the time the publisher drops the message. ## Two Ends and a Store The consumer is in the delivery operations context: it opens a delivery record for every message that arrives. Each message is assumed to cost three work units. **This is a model, not a measurement**; what is counted is the work unit count, no duration has been measured. The `keyed` option skips opening a record the second time it sees the same shipment code; it is measured at the end of the lesson. ```sh mkdir -p async ``` ```js // async/consumer.mjs — consumer: opens a delivery record for each message and spends 3 work units export const WORK_UNITS = 3; export function consumer(keyed = false) { const records = [], seen = new Set(); let units = 0; return { process(message) { units += WORK_UNITS; if (keyed && seen.has(message.code)) return; seen.add(message.code); records.push(message.code); }, units: () => units, records: () => records, }; } ``` The store exposes two operations. `enqueue` appends the message to the end and calls no consumer; `dequeue` gives the oldest message. The gap between these signatures is the whole of the style: enqueuer and dequeuer do not have to run at the same time. ```js // async/queue.mjs — message queue: the enqueuer does not wait, the dequeuer takes in order export function queue() { const pending = []; let enqueued = 0, dequeued = 0, peak = 0; return { enqueue(message) { pending.push(message); enqueued += 1; peak = Math.max(peak, pending.length); }, dequeue() { if (pending.length === 0) return null; dequeued += 1; return pending.shift(); }, length: () => pending.length, counts: () => ({ enqueued, dequeued, peak }), }; } ``` ## The Publisher Does Not Wait The first script passes twenty messages through both arrangements. In the synchronous arrangement the publisher calls the consumer directly; in the queue arrangement it drops the message. The consumer work run the moment the drop finishes is counted, the queue is then drained, and both arrangements are verified to produce the same record. In the second section, the producer processes five messages per round and the consumer three, and production stops after the fourth round. ```js // blocking.mjs — work run when the publisher drops the message, and messages pending in the queue import { consumer, WORK_UNITS } from "./async/consumer.mjs"; import { queue } from "./async/queue.mjs"; const MESSAGES = Array.from({ length: 20 }, (_, i) => ({ code: `G-${String(i + 1).padStart(2, "0")}` })); const sync = consumer(); for (const message of MESSAGES) sync.process(message); const q = queue(), later = consumer(); for (const message of MESSAGES) q.enqueue(message); const atDrop = { units: later.units(), pending: q.length() }; let dequeued; while ((dequeued = q.dequeue()) !== null) later.process(dequeued); const WIDTHS = [15, 25, 16, 18]; const ROWS = [ ["synchronous", MESSAGES.length, sync.units(), 0, sync.units()], ["queue", MESSAGES.length, atDrop.units, atDrop.pending, later.units()], ]; console.log("arrangement publisher calls work run at drop pending messages work run at end"); for (const [label, ...s] of ROWS) console.log(label.padEnd(13) + s.map((v, i) => String(v).padStart(WIDTHS[i])).join("")); console.log(`queue counts: ${JSON.stringify(q.counts())}, records equal = ${JSON.stringify(sync.records()) === JSON.stringify(later.records())}`); const PRODUCTION = [5, 5, 5, 5, 0, 0, 0, 0]; const CONSUMPTION = 3; const queue2 = queue(), consumer2 = consumer(); let sequence = 0; console.log("\nround produced consumed pending"); for (const [i, produced] of PRODUCTION.entries()) { for (let n = 0; n < produced; n += 1) { sequence += 1; queue2.enqueue({ code: `H-${sequence}` }); } let consumed = 0; for (let n = 0; n < CONSUMPTION; n += 1) { const message = queue2.dequeue(); if (message === null) break; consumer2.process(message); consumed += 1; } console.log(`${String(i + 1).padStart(3)}${String(produced).padStart(10)}${String(consumed).padStart(11)}${String(queue2.length()).padStart(10)}`); } console.log(`peak pending = ${queue2.counts().peak}, work units = ${consumer2.units()}, work units per message = ${WORK_UNITS}`); ``` ```sh node blocking.mjs ``` ``` arrangement publisher calls work run at drop pending messages work run at end synchronous 20 60 0 60 queue 20 0 20 60 queue counts: {"enqueued":20,"dequeued":20,"peak":20}, records equal = true round produced consumed pending 1 5 3 2 2 5 3 4 3 5 3 6 4 5 3 8 5 0 3 5 6 0 3 2 7 0 2 0 8 0 0 0 peak pending = 11, work units = 60, work units per message = 3 ``` ## Reading the Numbers The first table's last column is 60 in both arrangements: no work vanished, the same work got done, and both produced the same record. What changes is **when** the work happens. In the synchronous arrangement, sixty work units had already spent the publisher's own time by its twentieth call; in the queue arrangement, zero work units had run at the same moment, and twenty messages waited in the store. The pending-message count is the cost's name: the publisher escaped waiting, but unprocessed work now sits in the system and has to be held somewhere. The second table shows how this backlog behaves. While production runs faster than consumption, the pending count grows by two every round; after the fourth round's production, the queue reaches its peak, eleven. Once production stops, the queue drains and empties three rounds later. This is the style's core constraint: the queue **stores** the speed gap, it does not eliminate it. If production stays permanently faster than consumption, the pending count grows without bound; the queue only buys the publisher freedom from waiting while the gap is temporary. ## Order and Retry The second script asks two questions. The first is order: is the enqueue order the same as the processing order? It is measured separately for a single consumer and for two consumers working off the same queue's end. In the two-consumer arrangement, the workers' costs differ, and whichever worker frees up first takes the next message; finish order follows from that. **This is a model, not a measurement** — what is counted is the work unit and the finish order, no duration has been measured. The second question is retry: what happens if an unacknowledged message goes back onto the queue? ```js // order-and-retry.mjs — order with one and two consumers, delivery records opened when a message is redelivered import { consumer } from "./async/consumer.mjs"; import { queue } from "./async/queue.mjs"; const MESSAGES = Array.from({ length: 8 }, (_, i) => ({ code: `G-${i + 1}` })); const ORDER = MESSAGES.map((m) => m.code); const q = queue(), single = consumer(); for (const message of MESSAGES) q.enqueue(message); let dequeued; while ((dequeued = q.dequeue()) !== null) single.process(dequeued); const inversions = (sequence) => { const position = new Map(ORDER.map((code, i) => [code, i])); let count = 0; for (let a = 0; a < sequence.length; a += 1) for (let b = a + 1; b < sequence.length; b += 1) if (position.get(sequence[a]) > position.get(sequence[b])) count += 1; return count; }; const COST = [1, 4]; const freeAt = COST.map(() => 0); const finish = []; const queue2 = queue(); for (const message of MESSAGES) queue2.enqueue(message); while ((dequeued = queue2.dequeue()) !== null) { const j = freeAt[0] <= freeAt[1] ? 0 : 1; freeAt[j] += COST[j]; finish.push([dequeued.code, freeAt[j]]); } const finishOrder = [...finish].sort((a, b) => a[1] - b[1]).map(([code]) => code); console.log(`enqueue order : ${ORDER.join(" ")}`); console.log(`single consumer : ${single.records().join(" ")} inversions = ${inversions(single.records())}`); console.log(`two consumers : ${finishOrder.join(" ")} inversions = ${inversions(finishOrder)}`); for (const [label, keyed] of [["unkeyed", false], ["keyed", true]]) { const queue3 = queue(), consumer3 = consumer(keyed); for (const message of MESSAGES) queue3.enqueue(message); let count = 0; while ((dequeued = queue3.dequeue()) !== null) { consumer3.process(dequeued); count += 1; if (count === 3) queue3.enqueue(dequeued); } const counts = queue3.counts(); console.log(`${label.padEnd(11)} delivered = ${counts.dequeued}, distinct messages = ${new Set(MESSAGES.map((m) => m.code)).size}, opened delivery records = ${consumer3.records().length}, work units = ${consumer3.units()}`); } ``` ```sh node order-and-retry.mjs ``` ``` enqueue order : G-1 G-2 G-3 G-4 G-5 G-6 G-7 G-8 single consumer : G-1 G-2 G-3 G-4 G-5 G-6 G-7 G-8 inversions = 0 two consumers : G-1 G-3 G-4 G-2 G-5 G-6 G-8 G-7 inversions = 3 unkeyed delivered = 9, distinct messages = 8, opened delivery records = 9, work units = 27 keyed delivered = 9, distinct messages = 8, opened delivery records = 8, work units = 27 ``` For a single consumer, the inversion count is zero: order is preserved, a consequence of the store's `dequeue` signature. When a second consumer is added, the same queue hands out messages in the same order, but the finish order broke in three places. Order preservation is tied to having a single consumer; adding a second is not a performance decision, it is a decision to **give up the order guarantee**. The architectural consequence: work whose order matters and work that can be processed in parallel cannot go through the same queue. The last two lines are retry's bill. Eight distinct messages were delivered nine times. In the unkeyed consumer, opened delivery records numbered nine: two for the same shipment. In the keyed consumer it stayed at eight — the second delivery had no effect — but work units are 27 in both, since the message was taken and processed again regardless. This gives the style's constraint: the store puts the message in order, it does not guarantee **uniqueness**; that has to come from the consumer's own key. How a two-consumer design deviates against the same message arriving twice is measured separately in this course's Event-Driven Architecture lesson. ## Consuming versus Deleting The queue's `dequeue` operation removes the message from the store: consuming is deleting. The stream separates these operations — the record stays in place, and each consumer keeps its own **offset**. Offset, retention, and rereading were measured in the Streams versus Queues lesson of the Messaging topic. The question here is architectural: can a unit added afterward build its own history alone? ```js // async/log.mjs — stream: consuming does not delete, each consumer keeps its own offset export function log() { const entries = []; return { append: (message) => entries.push(message), read: (offset) => entries.slice(offset), length: () => entries.length }; } ``` ```js // history.mjs — queue empties once consumed, stream is read by offset: history for a unit added afterward import { queue } from "./async/queue.mjs"; import { log } from "./async/log.mjs"; import { consumer } from "./async/consumer.mjs"; const MESSAGES = Array.from({ length: 20 }, (_, i) => ({ code: `G-${i + 1}` })); const q = queue(), queueConsumer = consumer(); for (const message of MESSAGES) q.enqueue(message); let dequeued; while ((dequeued = q.dequeue()) !== null) queueConsumer.process(dequeued); const stream = log(), streamConsumer = consumer(); for (const message of MESSAGES) stream.append(message); let offset = 0; for (const message of stream.read(offset)) { streamConsumer.process(message); offset += 1; } const initialOffset = offset; const newcomer = consumer(); for (const message of stream.read(0)) newcomer.process(message); console.log(`queue: enqueued ${q.counts().enqueued}, pending after being consumed ${q.length()}`); console.log(` history a newcomer can read from the queue on its own = ${q.length()} records`); console.log(` the history is only obtained by exposing the consumer's records: ${queueConsumer.records().length} records, bindings created to another consumer = 1`); console.log(`stream: log length ${stream.length()}, first consumer's offset ${initialOffset}`); console.log(` history a newcomer reading from offset 0 can read = ${newcomer.records().length} records, bindings created to another consumer = 0`); console.log(` first consumer's offset ${initialOffset} -> ${offset}, log length ${stream.length()}`); ``` ```sh node history.mjs ``` ``` queue: enqueued 20, pending after being consumed 0 history a newcomer can read from the queue on its own = 0 records the history is only obtained by exposing the consumer's records: 20 records, bindings created to another consumer = 1 stream: log length 20, first consumer's offset 20 history a newcomer reading from offset 0 can read = 20 records, bindings created to another consumer = 0 first consumer's offset 20 -> 20, log length 20 ``` Once the queue has emptied, the history a newly added unit can read on its own is zero records. The only way to get the same information is for the existing consumer to expose its record — building a new binding between two consumers: the new unit now depends not on the publisher but on another consumer. In the stream the same number is twenty, and the binding created is zero; the new unit knows only the log. The second read left the first consumer's offset unchanged, and the log's length did not grow — consumer count does not grow the knowledge obligation. This is the choice criterion: if the work happens once and is done, a queue is enough; if units added afterward need to build their own state from history, a stream is needed — its cost is keeping the record around. ## Summary - The store separates the publisher from the consumer: a publisher dropping twenty messages spent 60 work units of its own time synchronously, spent 0 through the queue, and left 20 messages waiting in the store. - Total work is 60 in both arrangements and both produced the same record; the queue changes when work happens, not how much. - The queue stores the speed gap: with 5 produced and 3 consumed per round, the pending peak was 11, and the queue emptied three rounds after production stopped. - Order was preserved with a single consumer (0 inversions); adding a second consumer broke the finish order in three places — a decision to give up the order guarantee. - When eight messages were delivered nine times, the unkeyed consumer opened 9 delivery records and the keyed consumer 8; work units were 27 in both — the store guarantees order, not uniqueness. - In a queue, consuming is deleting: a unit added afterward can read 0 history records, versus 20 in a stream, which creates no binding to another consumer. ## Next Step This topic covered how units talk to each other. The same three things were measured across six styles: a unit's obligation to know the other side, the direction of connection, and the data crossing the boundary. Client and server, peers, model and view, pipe and filter, publisher and subscriber, the two ends of a queue — all compared on the same three numbers. But these measurements passed by one question: how many separate parts are these units deployed as? This lesson's queue, and the previous lesson's bus, both lived inside a single running program; the publish–subscribe arrangement needed no second process, one object was enough. Interaction style and part count are two independent decisions. The next topic asks the second one: how many separate deployment units does the software get published as, and what does that number change.