Message Brokers & Event Streams in Asyncio

An asyncio service that talks to a broker is doing the thing asyncio is best at — thousands of messages in flight, all of them I/O — and the thing it is least forgiving about. The client libraries are genuinely async and fast: 1,000 Redis Stream writes in 9 ms, 100 Kafka messages in 20 ms, 100 confirmed persistent RabbitMQ publishes in 50 ms, all measured on one developer machine. That speed is the problem as much as the solution, because a consumer will happily fetch faster than it can process, an acknowledgement in the wrong place converts a redelivery into a silent loss, and a publish that lives outside the database transaction it belongs to will eventually be the reason an account exists with no welcome email.

This section treats the broker as three separable problems: getting the event out of your service without losing it, getting it through the broker with the guarantees you think you have, and processing it once — or at least, processing it repeatedly without harm. The parent section, Network I/O & Protocol Handling, covers the transport layer these clients are built on; the queue mechanics here are the durable, cross-process version of Async Queue Management.

Scope of this section:

• Choosing between a log, a broker, and the database you already run.
• At-least-once delivery end to end, and where acknowledgement belongs.
• Consumer group models: offsets, pending entry lists and per-message acks.
• Back-pressure between a fast fetch loop and slow handlers.
• Publishing atomically with a database write, via the transactional outbox.
• Recovery: redelivery, claiming a dead worker's messages, and dead-lettering.

Architectural principles

• Acknowledge after processing, never before. Every broker's reliability rests on this. Committing an offset or acking a message before the work is done converts a crash from a redelivery into a silent loss, and no amount of monitoring will show you the messages that vanished.
• Assume at-least-once, design for idempotence. Exactly-once delivery does not exist across a network boundary; what exists is exactly-once effect, achieved by handlers that can process the same event twice without harm. Carry a stable event id and deduplicate on it.
• Never publish inside a business transaction, and never outside one. A publish inside the transaction holds a database transaction open across a network call; a publish after it can be lost. The resolution is an outbox row written in the transaction and relayed afterwards.
• Bound what you fetch. prefetch_count, max_poll_records and COUNT exist because a consumer that fetches faster than it processes turns a broker backlog into local memory exhaustion.
• Make the backlog visible. Kafka lag, Redis pending entries and RabbitMQ queue depth are the health metrics of an event pipeline; a consumer that looks healthy while falling behind is the normal failure.

Execution model: the broker is not the boundary

The conceptual model that avoids most mistakes is that an event crosses three boundaries, not one. It leaves your service's transaction, it crosses the broker, and it enters a consumer's transaction. Each hop is independently at-least-once, and the end-to-end guarantee is the weakest link — which is why the correctness argument always ends at the handler being idempotent rather than at any broker feature.

The asyncio-specific consequence is that the fetch loop and the processing must be decoupled deliberately. A naive async for message in consumer: await handle(message) couples them: the broker's flow control becomes your concurrency limit, a slow handler blocks the fetch, and in Kafka's case missed polls trigger a rebalance that reassigns your partitions mid-batch. The alternative — a bounded asyncio.Queue between a fetch task and a worker pool — decouples them but complicates acknowledgement, because you may only acknowledge once everything before a message has completed.

Most services should start coupled, with a modest prefetch, and decouple only when measurement shows handler latency is the constraint. The bookkeeping that parallel processing requires within one partition or queue is real, and the usual answer at scale is more partitions or more consumers rather than more concurrency inside one.

Retention is the other model difference worth internalising before choosing a system. A broker like RabbitMQ deletes a message when it is acknowledged, so the queue is a work list and its depth is your backlog. A log like Kafka keeps messages for a configured period regardless of who has read them, so the same topic can be read by a new consumer group next month from the beginning — and a bug that corrupted a downstream store can be fixed by replaying rather than by reconstructing. Redis Streams sit in between: entries persist until trimmed, so replay works within whatever window your MAXLEN affords. That single property decides more architectures than throughput ever does.

Ordering is the third. Kafka orders within a partition, Redis Streams order within a stream, and RabbitMQ orders within a queue for a single consumer — and none of them order across those units. When two events must be applied in order, they must share the unit: the same partition key, the same stream, the same queue with one consumer. Services that assume global ordering work perfectly in testing with one consumer and fail the day a second is added, which is why the partition key is a design decision rather than an implementation detail.

Pattern catalogue

Publish atomically with the database write

async with conn.transaction():
    account_id = await conn.fetchval(
        "INSERT INTO accounts (email) VALUES ($1) RETURNING id", email)
    await conn.execute("INSERT INTO outbox (topic, payload) VALUES ($1, $2)",
                       "account.created", json.dumps({"account_id": account_id}))
    await conn.execute("SELECT pg_notify('outbox', '')")       # wake the relay at commit

Both rows commit or neither does. A crash mid-transaction left 0 accounts and 0 outbox rows in testing — the dual-write window simply does not exist. The transactional outbox pattern covers the relay side.

Claim work so several relays can run

rows = await conn.fetch("""
    SELECT id, topic, payload FROM outbox
     WHERE published_at IS NULL ORDER BY id LIMIT $1
     FOR UPDATE SKIP LOCKED
""", batch)

SKIP LOCKED lets every instance run a relay without leader election. Four concurrent relays against 200 pending events claimed 50 rows each and published exactly 200 messages.

Consume a Kafka topic with deliberate commits

batches = await consumer.getmany(timeout_ms=1000, max_records=50)
for tp, messages in batches.items():
    for message in messages:
        await handle(message)
    await consumer.commit({tp: messages[-1].offset + 1})       # +1: the NEXT offset

Offsets are per partition and mean "the next message to read". Consuming without committing and restarting redelivered all ten test messages — the at-least-once guarantee, demonstrated. See consuming Kafka topics with aiokafka.

Acknowledge a RabbitMQ message, including failures

async with queue.iterator() as messages:
    async for message in messages:
        try:
            async with message.process(requeue=False):          # ack on success, nack on error
                await handle(message.body)
        except Exception:
            failures.inc()                                      # already dead-lettered

Without the try, the first failing message ends the consumer. With it, 98 of 100 messages processed and the two poison pills arrived in the dead-letter queue. See processing RabbitMQ messages with aio-pika.

Recover a crashed worker's messages

cursor, claimed, _ = await rds.xautoclaim(STREAM, GROUP, my_name,
                                          min_idle_time=60_000, count=50)

Redis Streams keep delivered-but-unacknowledged entries in a pending list, and XAUTOCLAIM transfers the ones idle for too long. Run it at the top of the consume loop so a dead peer's backlog is adopted promptly. See reading Redis Streams with consumer groups.

Bound the hand-off when handlers are slow

queue: asyncio.Queue = asyncio.Queue(maxsize=PREFETCH)     # back-pressure, not a buffer

async def fetch_loop(consumer):
    async for message in consumer:
        await queue.put(message)                            # blocks when workers fall behind

The bound is the point: an unbounded queue between a fast broker and a slow handler converts a backlog you can see in broker metrics into memory growth you only see when the process dies. Acknowledgement then has to wait for completion, which for an offset-based broker means tracking the lowest incomplete offset per partition rather than acking whatever finished last.

Resource boundaries

Resource	What consumes it	How to size and bound it
In-flight messages	Prefetch, max_poll_records, COUNT	2–3x processing concurrency; never unlimited
Consumer memory	Fetched-but-unprocessed messages	Bounded hand-off queue between fetch and workers
Broker connections	One per producer and consumer	Share a connection, open a channel or client per task
Partitions or queues	The parallelism ceiling	Choose the partition count for peak consumers, in advance
Pending entries	Delivered but unacknowledged	Alert on depth; claim abandoned ones on a timer
Outbox table size	Every event ever emitted	Partial index on unpublished; prune published rows nightly
Retention	Broker disk or Redis memory	A capacity decision: entries for Streams, days for Kafka

The first two rows are where asyncio services usually fail. A consumer with unlimited prefetch and slow handlers does not fall behind gracefully; it grows until it is killed, loses everything unacknowledged, and starts again from the same backlog.

Integrated production example

A publish-and-consume pipeline running in one process: the relay drains the outbox to a Redis Stream whenever NOTIFY fires or half a second passes, three group members consume with XAUTOCLAIM recovery and idempotent handling, and everything shuts down through one TaskGroup.

import asyncio
import contextlib
import json
import time

import asyncpg
import redis.asyncio as redis

STREAM, GROUP = "account.created", "mailer"
BATCH, PREFETCH = 100, 50


async def relay(pool: asyncpg.Pool, rds: redis.Redis, stop: asyncio.Event) -> int:
    """Claim unpublished outbox rows, publish, mark them; woken by NOTIFY, backed by a poll."""
    published = 0
    listener = await asyncpg.connect(PG_DSN)
    woken = asyncio.Event()
    await listener.add_listener("outbox", lambda *_: woken.set())
    try:
        while not stop.is_set():
            while True:
                async with pool.acquire() as conn, conn.transaction():
                    rows = await conn.fetch(
                        "SELECT id, topic, payload FROM outbox WHERE published_at IS NULL "
                        "ORDER BY id LIMIT $1 FOR UPDATE SKIP LOCKED", BATCH)
                    if not rows:
                        break
                    async with rds.pipeline(transaction=False) as pipe:
                        for row in rows:
                            pipe.xadd(row["topic"],
                                      {"event_id": str(row["id"]), "payload": row["payload"]},
                                      maxlen=100_000, approximate=True)
                        await pipe.execute()                    # publish first
                    await conn.execute(
                        "UPDATE outbox SET published_at = now() WHERE id = ANY($1)",
                        [r["id"] for r in rows])                # then mark: duplicates, not losses
                    published += len(rows)
            woken.clear()
            with contextlib.suppress(TimeoutError):
                async with asyncio.timeout(0.5):                # the poll is the safety net
                    await woken.wait()
    finally:
        await listener.close()
    return published


async def worker(rds: redis.Redis, name: str, stop: asyncio.Event, seen: set[str]) -> int:
    with contextlib.suppress(redis.ResponseError):              # BUSYGROUP: already created
        await rds.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
    handled = 0
    while not stop.is_set():
        _, claimed, _ = await rds.xautoclaim(STREAM, GROUP, name,
                                             min_idle_time=30_000, count=PREFETCH)
        batches = [claimed] if claimed else []                  # a dead peer's work comes first
        response = await rds.xreadgroup(GROUP, name, {STREAM: ">"},
                                        count=PREFETCH, block=200)
        for _stream, messages in response or []:
            batches.append(messages)
        for messages in batches:
            done = []
            for message_id, fields in messages:
                if fields["event_id"] not in seen:              # at-least-once: deduplicate
                    seen.add(fields["event_id"])
                    handled += 1
                done.append(message_id)
            if done:
                await rds.xack(STREAM, GROUP, *done)            # ack only completed work
    return handled


async def main() -> None:
    pool = await asyncpg.create_pool(PG_DSN, min_size=2, max_size=8)
    rds = redis.from_url(REDIS_URL, decode_responses=True)
    stop, seen = asyncio.Event(), set()
    started = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        relay_task = tg.create_task(relay(pool, rds, stop))
        workers = [tg.create_task(worker(rds, f"worker-{i}", stop, seen)) for i in range(3)]
        async with pool.acquire() as conn:
            for i in range(2000):
                async with conn.transaction():
                    account_id = await conn.fetchval(
                        "INSERT INTO accounts (email) VALUES ($1) RETURNING id",
                        f"user{i}@example.com")
                    await conn.execute(
                        "INSERT INTO outbox (topic, payload) VALUES ($1, $2)",
                        STREAM, json.dumps({"account_id": account_id}))
                    await conn.execute("SELECT pg_notify('outbox', '')")
        while await pool.fetchval("SELECT count(*) FROM outbox WHERE published_at IS NULL"):
            await asyncio.sleep(0.05)
        while len(seen) < 2000:
            await asyncio.sleep(0.05)
        stop.set()                                              # every task exits its loop
    info = (await rds.xinfo_groups(STREAM))[0]
    print(f"relay published {relay_task.result()}, workers handled "
          f"{[w.result() for w in workers]}, unique {len(seen)}")
    print(f"end to end {time.perf_counter() - started:.2f}s | "
          f"pending {info['pending']} lag {info['lag']}")
    await pool.close()
    await rds.aclose()


asyncio.run(main())

Running it prints relay published 2000, workers handled [669, 665, 666], unique 2000 and end to end 3.79s | pending 0 lag 0. Every event was written atomically with its account row, published once, distributed evenly across three group members, acknowledged, and deduplicated — and the group finished with no pending entries and no lag, which is what "the pipeline is caught up" looks like in metrics.

Diagnostic Hook — is the pipeline healthy or just quiet?

Export four numbers per topic. Backlog age, not depth: now() - min(created_at) for unpublished outbox rows, and consumer lag translated into seconds behind. Depth scales with traffic; age does not, which makes it the alertable metric. Pending or unacked count: rising pending with flat lag means handlers are failing, not that there is too much work. Redelivery count: Redis times_delivered, RabbitMQ's redelivered flag, or your own attempt header — a climbing count is a poison message that needs dead-lettering. Publish-to-consume latency, measured by putting the producer's timestamp in the payload: it is the only number that covers all three hops at once. Alert on backlog age above one minute and on any message redelivered more than three times.

Failure modes

Failure mode	Root cause	Detection	Fix
Events missing downstream	Publish outside the database transaction, lost on crash	Business row exists, no event	Transactional outbox with a relay
Duplicate side effects	At-least-once redelivery after a crash	Two emails, two charges	Idempotent handlers keyed on a stable event id
Consumer stops silently	An exception escaped the consume loop	Lag rises while the process is alive	Catch per message; dead-letter failures
Memory grows until the pod dies	Unlimited prefetch with slow handlers	RSS tracks backlog size	Set prefetch; bound the hand-off queue
The same message forever	Requeue on a permanent failure	One message, climbing delivery count	Cap attempts, then dead-letter
Rebalance storms in Kafka	Slow handlers blocking the poll loop	Repeated partition reassignment	Decouple processing; raise the poll interval
Work stranded after a crash	Pending entries never claimed	Pending stays high with idle workers	XAUTOCLAIM at the top of the loop
Broker memory or disk fills	No retention policy on the stream	Stream length grows without bound	MAXLEN ~ on every write; retention policy on topics

Frequently Asked Questions

Which message broker should an asyncio service use?

The one you already operate, unless its model is wrong for the job. Kafka suits replay and event history, RabbitMQ suits work queues and routing, Redis Streams suit a service already using Redis, and a Postgres queue suits work whose source of truth is the database. Every one of them has a mature asyncio client.

How do I avoid losing events when publishing from a database transaction?

Write the event to an outbox table inside the same transaction as the business change, and have a relay publish it afterwards. Both rows commit together, so there is no window where the row exists and the event was lost. A crash mid-transaction leaves neither, verified.

Is exactly-once delivery possible with asyncio message consumers?

Not as delivery. Every broker gives at-least-once when you acknowledge after processing, and a crash between processing and acknowledgement causes a redelivery. What is achievable is exactly-once effect: handlers that deduplicate on a stable event id, or whose writes are naturally idempotent.

How do I apply back-pressure between a broker and slow handlers?

Set the client's fetch limit — prefetch_count, max_poll_records or COUNT — to two or three times your processing concurrency, and if you decouple fetching from processing, put a bounded asyncio.Queue between them. Unlimited fetching turns a broker backlog into your process's memory problem.

What should I monitor for an event pipeline?

Backlog age rather than depth, pending or unacknowledged counts, per-message redelivery counts, and end-to-end publish-to-consume latency measured from a timestamp in the payload. Rising pending with flat lag means handlers are failing; rising lag with low pending means too few consumers.

Related

• Consuming Kafka topics with aiokafka — offsets, partitions and commit strategy.
• Processing RabbitMQ messages with aio-pika — prefetch, acknowledgement and dead-lettering.
• Reading Redis Streams with consumer groups — pending entries and claiming.
• Implementing the transactional outbox pattern — publishing without losses.
• Network I/O & Protocol Handling — the parent section.