Batch Send
Send multiple KubeMQ Queue messages in a single batch with the Java SDK to reduce round-trips and boost throughput.
Which to use
This page is about grouping a known, finite set of messages into one logical unit — tracking per-item success/failure and aggregate throughput for that batch. For sending a continuous, high-volume stream of messages over the SDK's shared upstream connection, see Stream Send.
Overview
Batch sending is the practice of publishing many related queue messages together — importing records, fanning out a set of jobs, replaying a backlog — instead of firing off one-off sends scattered through your code. Grouping the work matters because each sendQueueMessage call is an independent, synchronous round trip to the broker; treating a set of messages as one logical unit gives you a single place to track success/failure counts and throughput.
The sample builds each QueueMessage with the QueueMessage.builder()...build() fluent API, attaching tags (batch_id, sequence) that consumers can use for routing or deduplication, then calls client.sendQueueMessage(msg) once per message in a loop, tallying result.isError() and timing the loop to report throughput.
Gotchas: each sendQueueMessage call confirms storage individually, so a failure partway through doesn't roll back messages already accepted — you get partial success, not all-or-nothing; looping synchronous single sends bounds throughput by round-trip latency times message count, so for large or continuous publishing a true batch call or a persistent upstream stream scales far better; and tags add per-message overhead, so keep them small and structured.
Prerequisites
- KubeMQ server running on
localhost:50000 - Java SDK installed (
implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1'(Gradle) or Maven dependency from Getting Started)
Code
package io.kubemq.example.queues;
import io.kubemq.sdk.common.ServerInfo;
import io.kubemq.sdk.queues.*;
import java.util.*;
public class BatchSendExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-batch-send-client";
private static final String CHANNEL = "java-queues.batch-send";
public static void main(String[] args) {
// Create a queues client connected to the KubeMQ server
QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build();
// Verify connection to the server
ServerInfo info = client.ping();
System.out.println("Connected to: " + info.getHost());
// Create the queue channel
client.createQueuesChannel(CHANNEL);
// Send a batch of messages
int batchSize = 10;
int success = 0;
long start = System.currentTimeMillis();
for (int i = 1; i <= batchSize; i++) {
Map<String, String> tags = new HashMap<>();
tags.put("batch_id", UUID.randomUUID().toString());
tags.put("sequence", String.valueOf(i));
QueueMessage msg = QueueMessage.builder()
.id(UUID.randomUUID().toString()).channel(CHANNEL)
.body(("Batch message #" + i).getBytes()).tags(tags).build();
QueueSendResult result = client.sendQueueMessage(msg);
if (!result.isError()) { success++; }
}
long elapsed = System.currentTimeMillis() - start;
System.out.println("Batch complete: " + success + "/" + batchSize + " in " + elapsed + "ms");
System.out.println("Throughput: " + (batchSize * 1000 / Math.max(elapsed, 1)) + " msg/sec");
// Clean up resources
client.deleteQueuesChannel(CHANNEL);
client.close();
}
}
How It Works
sendQueueMessageis called in a tight loop; each call confirms storage individually sosuccesstracks only broker-accepted messages.- Tags (
batch_id,sequence) are attached to each message asMap<String, String>and stored alongside the body; consumers can use them for routing or deduplication. - Elapsed time and throughput are measured around the full batch loop to show the cost of synchronous per-message sends vs. the alternatives (stream or batch API).
Related
Was this page helpful?