Rabbit Batch Subscriber#

FastStream does not provide a built-in batch consumer API for RabbitMQ. Hiding message accumulation behind a special subscriber option would obscure how prefetch, manual acknowledgement, and buffering actually work — so this page shows an application-level pattern instead.

In the example below, FastStream still delivers messages one by one. Your code buffers them in memory and flushes when the batch reaches BATCH_SIZE. For sparse traffic, add a timeout (or shutdown drain) so a partial batch does not sit unacked forever.

Warning

This pattern keeps unacknowledged messages in process memory until a batch completes. Processing and acknowledgement are not atomic: if the worker stops after your business logic but before every ack, RabbitMQ will redeliver those messages. Batch handling must be idempotent — safe to run more than once for the same payload.

Note

The buffer lives in a single worker process. Running multiple app instances does not build one shared batch — each instance buffers its own deliveries.

Batch Subscriber Pattern#

Let's dive into example code:

import asyncio
from faststream import AckPolicy, FastStream, Logger
from faststream.rabbit import Channel, RabbitBroker, RabbitMessage, RabbitQueue

broker = RabbitBroker()
app = FastStream(broker)
queue = RabbitQueue("example_queue")
BATCH_SIZE = 10
lock = asyncio.Lock()
batch: list[tuple[dict, RabbitMessage]] = []

Broker setup, BATCH_SIZE, and shared state: an asyncio.Lock, the in-memory batch buffer.

def take_batch() -> list[tuple[dict, RabbitMessage]]:
    """Detach the current batch while the lock is held."""
    items = batch.copy()
    batch.clear()
    return items


async def flush(items: list[tuple[dict, RabbitMessage]], logger: Logger) -> None:
    payloads = [msg for msg, _ in items]
    processed.append(payloads)
    logger.info("Processing batch of %s messages", len(payloads))
    # your batch work here
    for _, raw in items:
        await raw.ack()

take_batch() — detaches the current buffer under the lock (copy + clear)
flush() — runs your batch work on the detached items, then acknowledges each message

@broker.subscriber(
    queue,
    channel=Channel(prefetch_count=BATCH_SIZE * 2),
    ack_policy=AckPolicy.MANUAL,
)
async def handle(message: dict, raw_message: RabbitMessage, logger: Logger) -> None:
    items: list[tuple[dict, RabbitMessage]] | None = None
    async with lock:
        batch.append((message, raw_message))
        if len(batch) >= BATCH_SIZE:
            items = take_batch()  # detach under the lock
    if items is not None:
        await flush(items, logger)  # process outside the lock

Under the lock the handler appends the message, and when the buffer reaches BATCH_SIZE it calls take_batch(). After releasing the lock it calls flush() on that detached batch — so concurrent deliveries cannot grow a batch past BATCH_SIZE.

Note

Use AckPolicy.MANUAL and limit in-flight messages with Channel(prefetch_count=...) on @broker.subscriber. Prefetch is your main defense against unbounded memory growth while messages wait for a full batch.

Full Example

import asyncio
from faststream import AckPolicy, FastStream, Logger
from faststream.rabbit import Channel, RabbitBroker, RabbitMessage, RabbitQueue

broker = RabbitBroker()
app = FastStream(broker)
queue = RabbitQueue("example_queue")
BATCH_SIZE = 10
lock = asyncio.Lock()
batch: list[tuple[dict, RabbitMessage]] = []
processed: list[list[dict]] = []  # used in docs tests


def take_batch() -> list[tuple[dict, RabbitMessage]]:
    """Detach the current batch while the lock is held."""
    items = batch.copy()
    batch.clear()
    return items


async def flush(items: list[tuple[dict, RabbitMessage]], logger: Logger) -> None:
    payloads = [msg for msg, _ in items]
    processed.append(payloads)
    logger.info("Processing batch of %s messages", len(payloads))
    # your batch work here
    for _, raw in items:
        await raw.ack()


@broker.subscriber(
    queue,
    channel=Channel(prefetch_count=BATCH_SIZE * 2),
    ack_policy=AckPolicy.MANUAL,
)
async def handle(message: dict, raw_message: RabbitMessage, logger: Logger) -> None:
    items: list[tuple[dict, RabbitMessage]] | None = None
    async with lock:
        batch.append((message, raw_message))
        if len(batch) >= BATCH_SIZE:
            items = take_batch()  # detach under the lock
    if items is not None:
        await flush(items, logger)  # process outside the lock