Batch Send

Overview

Batch send groups several queue messages into one call instead of sending them one at a time. Reach for it when publishing many related items together — importing records, fanning out a set of jobs, replaying a backlog — since sending each message individually pays a full round trip per message, while batching amortizes that cost across the whole set.

It works by building a list of QueueMessage objects, then passing the list to client.send_queue_messages_batch(messages) in a single gRPC call. The broker persists each message independently and returns one QueueSendResult per message, in the same order as the input.

Gotchas: batching isn't atomic — the broker can accept some messages and reject others in the same call, so always check each result rather than trusting an overall success; a batch is still one bounded request, so it doesn't help continuous, open-ended publishing (use a stream-based send for that); and very large batches raise the size and latency of that single call, so there's a practical ceiling before splitting into multiple batches pays off.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Batch send — send multiple queue messages in a single batch."""

from __future__ import annotations

import asyncio

from kubemq.queues import AsyncClient as AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-queues-batch-send-client",
    ) as client:
        messages = [
            QueueMessage(
                channel="python-queues.batch-send",
                body=f"Batch msg #{i + 1}".encode(),
            )
            for i in range(5)
        ]
        results = await client.send_queue_messages_batch(messages)
        for r in results:
            print(f"Sent: id={r.id}, sent_at={r.sent_at}")

        poll = await client.receive_queue_messages(
            channel="python-queues.batch-send",
            max_messages=10,
            wait_timeout_seconds=5,
            auto_ack=True,
        )
        for msg in poll.messages:
            print(f"Received: {msg.body.decode('utf-8')}")


if __name__ == "__main__":
    asyncio.run(main())

How It Works

• send_queue_messages_batch sends a list of QueueMessage objects in a single gRPC call, returning one QueueSendResult per message.
• Batching reduces per-message gRPC overhead; the broker persists each message atomically regardless of the others in the batch.
• The receive call uses auto_ack=True to automatically acknowledge all polled messages, appropriate here for a simple drain.
• Individual send results allow per-message error detection within the batch if the broker rejects specific messages.

Related

• Pattern overview
• Python SDK Reference
• Send & Receive
• Ack All