Send & Receive
Send and receive messages on a KubeMQ Queue channel with the Java SDK in a basic producer-consumer workflow.
Overview
Queue send/receive is the foundational operation for guaranteed-delivery, point-to-point messaging: you reach for it whenever work needs to survive past the moment it's created and be handled by exactly one consumer, not broadcast to every subscriber. Unlike pub/sub, a queued message sits durably on the broker until something pulls it, so the producer and consumer never need to be online at the same time — a slow or offline worker adds latency, it doesn't drop the message.
This tutorial builds the smallest possible version of that round trip: client.sendQueueMessage(message) enqueues a message on a channel, and client.receiveQueueMessages(pollRequest) pulls it back within a bounded pollWaitTimeoutInSeconds. Settlement here is manual — each QueueMessageReceived must be explicitly confirmed with msg.ack() once your handler finishes; nothing is removed from the queue automatically.
Gotchas: if your handler crashes before calling msg.ack(), the message stays in the queue and becomes available for redelivery again once its visibility timeout elapses — write handlers that tolerate seeing the same message twice. Calling receiveQueueMessages against an empty queue isn't an error; it just blocks up to pollWaitTimeoutInSeconds and returns an empty list. And pollMaxMessages caps the batch size per call, so don't assume one poll drains the whole queue.
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.queues.*;
public class SendReceiveExample {
// TODO: Replace with your KubeMQ server address
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-send-receive-client";
private static final String CHANNEL = "java-queues.send-receive";
public static void main(String[] args) {
// Create a queues client connected to the KubeMQ server
try (QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build()) {
// Build and send a queue message
QueueMessage message = QueueMessage.builder()
.channel(CHANNEL).body("Hello KubeMQ Queue!".getBytes()).metadata("simple-example").build();
// Send the message to the queue
QueueSendResult sendResult = client.sendQueueMessage(message);
System.out.println("Message sent - ID: " + sendResult.getId() + ", Error: " + sendResult.isError());
// Poll to receive messages from the queue
QueuesPollRequest pollRequest = QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(1).pollWaitTimeoutInSeconds(10).build();
// Receive messages from the queue
QueuesPollResponse response = client.receiveQueueMessages(pollRequest);
if (response.isError()) {
System.err.println("Receive error: " + response.getError());
} else {
// Handle each received message and acknowledge it
response.getMessages().forEach(msg -> {
System.out.println("Received - ID: " + msg.getId() + ", Body: " + new String(msg.getBody()));
msg.ack();
});
}
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
}
}
}
// Expected output:
// Message sent - ID: <message-id>, Error: false
// Received - ID: <message-id>, Body: Hello KubeMQ Queue!
How It Works
client.sendQueueMessage(message)returnsQueueSendResultwith an ID and error flag; a non-error result confirms the broker accepted and stored the message.QueuesPollRequestcontrols the pull window:pollMaxMessages(1)limits the batch size andpollWaitTimeoutInSeconds(10)is the long-poll timeout if the queue is empty.client.receiveQueueMessages(pollRequest)blocks up to the timeout and returns aQueuesPollResponse; eachQueueMessageReceivedin the list must be explicitlyack()ed to remove it from the queue.
Related
Was this page helpful?