Skip to content

Fanning Out a Queue to Multiple Consumer Groups

An ingestion service receives change events and needs three things done with each one: index it for search, write an audit record, and update metrics. The obvious implementation puts events on an asyncio.Queue and starts three consumers — and then each event is delivered to exactly one of them, because a queue distributes work rather than broadcasting it. The second attempt calls the three handlers inline, which couples them: the audit writer's 20 ms of I/O now sets the pace for indexing, and one handler's failure loses the event for the others. What this needs is fan-out: every subscriber gets its own queue, consumes at its own speed, and fails independently. This guide builds that in-process bus, chooses a policy for subscribers that fall behind, handles subscribe and unsubscribe under load, exports lag per subscriber, and adds a replay buffer for late joiners.

Prerequisites

Distribution versus fan-out 2 columns contrasting one shared queue, queue per subscriber. Distribution versus fan-out one shared queue work distribution each item to one consumer right for worker pools consumers share the load queue per subscriber fan-out each item to every consumer right for independent work consumers fail separately A queue hands work out; a bus hands copies out.

1. Understand why one queue cannot broadcast

asyncio.Queue.get() removes an item. With several consumers on one queue, each item goes to whichever consumer happened to be waiting — that is work distribution, which is the right tool for a worker pool and the wrong one for "everyone must see this".

import asyncio
import collections


async def one_queue_three_consumers() -> dict[str, int]:
    queue: asyncio.Queue[int] = asyncio.Queue()
    counts: collections.Counter[str] = collections.Counter()

    async def consumer(name: str) -> None:
        while True:
            item = await queue.get()
            counts[name] += 1
            queue.task_done()

    consumers = [asyncio.create_task(consumer(n)) for n in ("indexer", "audit", "metrics")]
    for i in range(30):
        queue.put_nowait(i)
    await queue.join()
    for task in consumers:
        task.cancel()
    return dict(counts)


print(asyncio.run(one_queue_three_consumers()), "-> 30 items split, not broadcast")

The counts sum to 30 across the three consumers rather than being 30 each — in this run a single consumer took all 30, because get() on a non-empty queue returns without suspending, so the first consumer drained the queue before the others ran. The distribution between consumers is incidental; what matters is that each item was delivered exactly once. Every consumer group that must see every item therefore needs its own queue, and something has to put a copy of each item into all of them.

Verify: the counts add up to 30 rather than to 90, whichever consumer received them.

2. Give every subscriber its own bounded queue

The bus keeps a set of subscriptions, each with a bounded queue. Publishing is synchronous and never awaits a subscriber, so one slow consumer cannot slow the producer or the other consumers.

import asyncio
import contextlib
from dataclasses import dataclass


@dataclass(eq=False)                                        # identity hashing: usable in a set
class Subscription:
    name: str
    queue: asyncio.Queue
    dropped: int = 0
    received: int = 0


class FanOut:
    def __init__(self, maxsize: int = 64, policy: str = "drop_oldest") -> None:
        self.maxsize = maxsize
        self.policy = policy                                # drop_oldest | drop_newest
        self.subscribers: set[Subscription] = set()

    @contextlib.asynccontextmanager
    async def subscribe(self, name: str):
        sub = Subscription(name, asyncio.Queue(maxsize=self.maxsize))
        self.subscribers.add(sub)                           # no await: publish cannot interleave
        try:
            yield sub
        finally:
            self.subscribers.discard(sub)                   # unsubscribing is always safe

    def publish(self, item) -> None:
        for sub in self.subscribers:
            try:
                sub.queue.put_nowait(item)
            except asyncio.QueueFull:
                sub.dropped += 1
                if self.policy == "drop_oldest":
                    with contextlib.suppress(asyncio.QueueEmpty):
                        sub.queue.get_nowait()              # make room for the newest item
                    sub.queue.put_nowait(item)

    def lag(self) -> dict[str, int]:
        return {sub.name: sub.queue.qsize() for sub in self.subscribers}

Adding to and removing from the subscriber set happens without any await, so publish() can never iterate a set that is being modified. Publishing a shared, immutable item is important too: subscribers receive the same object, so mutating it in one consumer would corrupt it for the others.

Verify: subscribing inside async with adds exactly one subscription, and leaving the block removes it even if the consumer raised.

3. Choose what happens when a subscriber falls behind

A bounded queue forces a decision: drop the oldest item, drop the newest, or apply backpressure to the producer. Fan-out to independent consumers almost always drops, because blocking the publisher would let the slowest subscriber dictate the whole pipeline.

import asyncio


async def consume(sub: Subscription, work_seconds: float, stop: asyncio.Event) -> None:
    while not stop.is_set():
        try:
            async with asyncio.timeout(0.05):
                item = await sub.queue.get()
        except TimeoutError:
            continue                                        # idle: check the stop flag
        await asyncio.sleep(work_seconds)
        sub.received += 1


async def main() -> None:
    bus = FanOut(maxsize=8, policy="drop_oldest")
    stop = asyncio.Event()
    async with bus.subscribe("indexer") as indexer, \
               bus.subscribe("audit") as audit, \
               bus.subscribe("metrics") as metrics:
        consumers = [asyncio.create_task(consume(sub, delay, stop))
                     for sub, delay in ((indexer, 0.0), (audit, 0.02), (metrics, 0.001))]
        for i in range(200):
            bus.publish({"id": i})                          # never awaits a subscriber
            await asyncio.sleep(0.001)
        print("lag while publishing:", bus.lag())
        await asyncio.sleep(0.05)
        stop.set()
        await asyncio.gather(*consumers)
        print({s.name: {"received": s.received, "dropped": s.dropped}
               for s in (indexer, audit, metrics)})


asyncio.run(main())

The fast indexer and metrics consumers received all 200 events with no drops; the audit consumer, which needs 20 ms per event, received 13 and dropped 181 while its queue sat at its limit of 8. That is the correct outcome if audit records may be sampled — and the wrong one if they may not, in which case audit belongs behind a durable queue rather than an in-memory bus, or the producer must be slowed deliberately for that subscriber.

Verify: fast subscribers report zero drops, the slow subscriber's queue stays at maxsize, and its drop count grows.

Delivery with one slow subscriber 4 bars comparing indexer received with the others. Delivery with one slow subscriber indexer received 200 metrics received 200 audit dropped 181 dropped audit received 13 Measured with the step 3 code: 200 events at 1 ms, subscriber queues bounded at 8. The slow subscriber loses events; nobody else is slowed down.

4. Export lag per subscriber and act on it

Drops are a symptom; lag is the leading indicator. Export queue depth and drop count per subscriber, and treat sustained lag as an alert on that consumer rather than on the bus.

import asyncio


class ObservedFanOut(FanOut):
    def __init__(self, maxsize: int = 64, policy: str = "drop_oldest",
                 lag_alert_ratio: float = 0.8) -> None:
        super().__init__(maxsize, policy)
        self.lag_alert_ratio = lag_alert_ratio

    def health(self) -> dict[str, dict]:
        report = {}
        for sub in self.subscribers:
            depth = sub.queue.qsize()
            report[sub.name] = {
                "lag": depth,
                "lag_ratio": round(depth / self.maxsize, 2),
                "dropped": sub.dropped,
                "alert": depth >= self.maxsize * self.lag_alert_ratio,
            }
        return report


async def main() -> None:
    bus = ObservedFanOut(maxsize=10)
    stop = asyncio.Event()
    async with bus.subscribe("fast") as fast, bus.subscribe("slow") as slow:
        consumers = [asyncio.create_task(consume(fast, 0.0, stop)),
                     asyncio.create_task(consume(slow, 0.03, stop))]
        for i in range(60):
            bus.publish(i)
            await asyncio.sleep(0.001)
        print(bus.health())
        stop.set()
        await asyncio.gather(*consumers)


asyncio.run(main())

A subscriber whose lag_ratio sits near one is about to start dropping; one whose lag oscillates is keeping up on average. Because each subscriber has its own queue, the alert names the consumer that is slow, which is the information an on-call engineer needs — unlike a single shared queue, where depth says only that "something" is behind.

Verify: the fast subscriber reports a lag near zero and no alert, and the slow one reports a lag ratio approaching one with alert: True.

5. Let late subscribers catch up with a replay buffer

Consumers that start after the producer — a debugging session, a newly enabled feature, a reconnecting WebSocket client — see nothing of what came before. A small ring buffer of recent items lets a new subscriber replay them before joining the live stream.

import asyncio
import collections
import contextlib


class ReplayFanOut(FanOut):
    def __init__(self, maxsize: int = 64, replay: int = 5) -> None:
        super().__init__(maxsize)
        self.history: collections.deque = collections.deque(maxlen=replay)

    def publish(self, item) -> None:
        self.history.append(item)                           # newest `replay` items retained
        super().publish(item)

    @contextlib.asynccontextmanager
    async def subscribe(self, name: str, replay: bool = True):
        async with super().subscribe(name) as sub:
            if replay:
                for item in self.history:                   # no await: the live stream cannot interleave
                    with contextlib.suppress(asyncio.QueueFull):
                        sub.queue.put_nowait(item)
            yield sub


async def main() -> None:
    bus = ReplayFanOut(maxsize=16, replay=3)
    for i in range(10):
        bus.publish(f"event-{i}")                           # nobody is listening yet
    async with bus.subscribe("late-joiner") as sub:
        bus.publish("event-10")
        received = [sub.queue.get_nowait() for _ in range(sub.queue.qsize())]
    print(received)                                          # last 3 old events, then the live one


asyncio.run(main())

The late joiner received event-7, event-8, event-9 from the buffer and then event-10 live, in order, because the replay happens synchronously inside subscribe(). Size the buffer for the gap you need to cover — a reconnect window, a deploy — not as a substitute for durable storage; for guaranteed delivery across restarts, put the events in a broker or a database as described in implementing the transactional outbox pattern in asyncio.

Verify: the late joiner's list starts with the last three historical events and ends with the live one.

What may this subscriber lose? A decision on What does this consumer do with events with 3 outcomes. What may this subscriber lose? What does this consumer do with events? samples or aggregates drop oldest bounded queue needs the latest value conflate to newest depth of one must not lose any durable queue not an in-memory bus The third branch is a storage decision, not a queue-size decision.

Verification

In-process fan-out is correct when:

  • Every subscriber sees every item it was subscribed for, rather than a share of them.
  • The publisher never awaits a subscriber: publish is synchronous and one slow consumer cannot slow the producer.
  • The drop policy is deliberate: each subscriber's queue bound and policy match what that consumer may lose.
  • Subscribe and unsubscribe are safe under load: they happen without awaits, and leaving the block always removes the subscription.
  • Lag and drops are exported per subscriber, and alerts name the slow consumer.

Pitfalls & edge cases

  • Mutating shared items. Subscribers receive the same object; one consumer editing it changes what the others see. Publish immutable payloads or copies.
  • Unbounded subscriber queues. An unbounded queue turns a slow consumer into a memory leak instead of a drop counter. Always bound, then choose a policy.
  • Awaiting inside publish. Any await in the publish path lets one subscriber's backpressure reach the producer; use put_nowait and handle QueueFull.
  • Losing subscriptions on consumer errors. If a consumer task dies, its queue keeps filling and dropping silently. Supervise consumers and unsubscribe when they stop.
  • Fan-out across processes. In-memory subscriptions are per process; in a multi-worker deployment each process needs its own subscription to the external broker topic.

Frequently Asked Questions

Can several asyncio consumers read the same items from one Queue?

No. Queue.get removes the item, so with several consumers each item goes to exactly one of them. That is work distribution. To give every consumer every item, keep one queue per consumer and have the producer put a copy of each item into all of them.

How do I handle a slow subscriber in an in-process fan-out?

Bound each subscriber's queue and decide what happens when it is full: drop the oldest item, drop the newest, or block the publisher. For independent consumers, dropping is usually right, since blocking lets the slowest subscriber set the pace for everyone; count drops per subscriber and alert on them.

How can a new subscriber see recent events it missed?

Keep a bounded ring buffer of recent items in the bus and, when a subscriber joins, copy those items into its queue before it starts receiving live ones, without awaiting in between. Size the buffer for the gap you need to cover; durable delivery needs a broker or database instead.

Is it safe to subscribe and unsubscribe while events are being published?

Yes, if both operations happen without an await between reading and modifying the subscriber collection, because the event loop cannot interleave them. Using an async context manager for subscriptions also guarantees that a consumer that raises or is cancelled removes its subscription.