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 messages with the queueMessage { ... } builder, then passing the list to client.sendQueueMessagesBatch(messages) in a single RPC call. The broker enqueues each message independently and returns one result per message — each carrying a messageId and isError — 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 isError on every result rather than trusting the batch as a whole; 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
• Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

package io.kubemq.sdk.examples.queues

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.queues.queueMessage
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-queues-batch-send"
private const val CHANNEL = "kotlin-queues.batch-send"

fun main() = runBlocking {
    val client = KubeMQClient.queues {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        // Batch send using the simple API
        val messages = (1..10).map { i ->
            queueMessage {
                channel = CHANNEL
                body = "Batch item #$i".toByteArray()
                metadata = "batch-job"
            }
        }

        println("Sending ${messages.size} messages in batch...")
        val start = System.currentTimeMillis()
        val results = client.sendQueueMessagesBatch(messages)
        val elapsed = System.currentTimeMillis() - start

        results.forEach { r ->
            println("  Sent: ${r.messageId}, error=${r.isError}")
        }
        println("\n${results.size} messages sent in ${elapsed}ms")
        println("Throughput: ${results.size * 1000.0 / elapsed} msg/s")

        // Cleanup: consume the messages
        val cleanup = client.receiveQueuesMessages {
            channel = CHANNEL
            maxItems = 10
            waitTimeoutMs = 3000
            autoAck = true
        }
        println("Cleanup: consumed ${cleanup.messages.size} messages.")

        println("Done.")
    }
}

How It Works

• sendQueueMessagesBatch(messages) sends a list of messages in a single RPC for higher throughput.
• Each result includes messageId and isError status for individual message tracking.
• Batch sending reduces network round trips compared to individual sendQueuesMessage() calls.

Related

• Send & Receive
• Stream Send
• Queues Reference