Stream Send
Publish persistent KubeMQ Events Store messages at high throughput using the streaming send API in the Java SDK.
Overview
Stream send covers publishing a batch of persistent events back-to-back, without pausing your producer between messages. A single publishEventStore call is fine for one-off writes, but if you're bulk-loading history, replicating a firehose of records, or backfilling an Events Store channel, sending events in rapid succession turns network latency into your throughput ceiling instead of an app-level bottleneck.
Each publishEventStore(message) call still confirms storage synchronously, returning an EventSendResult whose isSent() flag tells you the broker persisted the event before the loop advances to the next one. Tracking elapsed time and successful sends across the batch gives you a real throughput number you can compare against other publishing strategies. Gotchas: each call blocks on its own confirmation, so this loop is latency-bound — very high fan-out workloads need concurrent sends rather than one tight loop; a single failed isSent() doesn't stop the batch, so you must check every result if you need all-or-nothing delivery; and don't treat printed throughput numbers as a benchmark without warming up the connection and running a realistic payload size first.
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.eventsstore;
import io.kubemq.sdk.common.ServerInfo;
import io.kubemq.sdk.pubsub.*;
import java.util.UUID;
/**
* Stream Send Example (EventsStore)
*
* Demonstrates high-throughput streaming of events store messages.
*/
public class StreamSendExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-eventsstore-stream-send-client";
private static final String CHANNEL = "java-eventsstore.stream-send";
public static void main(String[] args) throws InterruptedException {
// Create a client connected to the KubeMQ server
PubSubClient client = PubSubClient.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 events store channel
client.createEventsStoreChannel(CHANNEL);
int messageCount = 10;
// Send multiple events store messages (high-throughput)
System.out.println("Sending " + messageCount + " events store messages via stream...\n");
long start = System.currentTimeMillis();
int successCount = 0;
for (int i = 1; i <= messageCount; i++) {
EventStoreMessage message = EventStoreMessage.builder()
.id(UUID.randomUUID().toString())
.channel(CHANNEL)
.body(("Stream store event #" + i).getBytes())
.metadata("stream-batch")
.build();
EventSendResult result = client.publishEventStore(message);
if (result.isSent()) {
successCount++;
}
}
long elapsed = System.currentTimeMillis() - start;
System.out.println("Sent " + successCount + "/" + messageCount + " events in " + elapsed + "ms");
System.out.println("Throughput: " + (messageCount * 1000 / Math.max(elapsed, 1)) + " msg/sec");
// Clean up resources
client.deleteEventsStoreChannel(CHANNEL);
client.close();
System.out.println("\nStream send example completed.");
}
}
How It Works
client.publishEventStore(message)returnsEventSendResultsynchronously; each call confirms storage on the broker before the loop advances.successCounttracks acknowledged sends; any broker-side rejection would decrement from the expected 10.- Elapsed time and throughput printed at the end measure the cost of synchronous acknowledgement per message — useful for comparing against plain Events publish or batching strategies.
Related
Was this page helpful?