Poll Mode
Pull KubeMQ Queue messages on demand with the Java SDK using poll mode for explicit, batch-controlled consumption.
Overview
Poll mode is a pull-based way to consume queue messages: the consumer decides exactly when to ask for work and how much, instead of holding an open stream the broker pushes into. That control matters for batch jobs, cron-triggered workers, and any consumer that only runs intermittently and would rather ask "is there anything for me?" than keep a subscription alive.
A single call to receiveQueueMessages sends a channel, pollMaxMessages, and pollWaitTimeoutInSeconds; the broker holds the request open as a long poll and returns once enough messages are available or the timeout elapses, so the call never spins on an empty queue. Each returned message is acknowledged individually with ack(), giving you a chance to skip one you can't process.
Gotchas: un-acked messages return to the queue only after the broker's visibility timeout, so a slow or crashed consumer can leave them invisible to others for a while; the timeout bounds latency, not throughput, so a small pollMaxMessages on a busy queue means many round trips; and any message beyond your batch size (5 queued but only 3 requested) simply waits for the next poll — it isn't lost.
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.queuesstream;
import io.kubemq.sdk.queues.*;
import java.util.UUID;
/**
* PollModeExample for Queues Stream
*/
public class PollModeExample {
private static final String ADDRESS = "localhost:50000";
private static final String CLIENT_ID = "java-queues-poll-mode-client";
private static final String CHANNEL = "java-queues.poll-mode";
public static void main(String[] args) {
// Create a client connected to the KubeMQ server
try (QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build()) {
// Create the queue channel
client.createQueuesChannel(CHANNEL);
// Send messages to the queue
for (int i = 1; i <= 5; i++) {
client.sendQueueMessage(QueueMessage.builder()
.channel(CHANNEL).body(("Poll msg " + i).getBytes()).build());
}
// Poll for messages in pull mode
System.out.println("=== Waiting Pull Mode ===\n");
QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
.channel(CHANNEL).pollMaxMessages(3).pollWaitTimeoutInSeconds(10).build());
// Process and acknowledge each received message
System.out.println("Received " + response.getMessages().size() + " messages:");
response.getMessages().forEach(msg -> {
System.out.println(" " + new String(msg.getBody()));
msg.ack();
});
// Clean up resources
client.deleteQueuesChannel(CHANNEL);
} catch (Exception e) {
System.err.println("Error: " + e.getMessage());
}
}
}
How It Works
pollMaxMessages(3)limits the batch to 3 even though 5 are queued; the remaining 2 stay in the queue until the next poll.pollWaitTimeoutInSeconds(10)is the server-side long-poll timeout; the call returns as soon as messages are available or after 10 seconds if the queue stays empty.- Each received message is
ack()ed individually inside theforEach; un-acked messages are returned to the queue after the broker's visibility timeout.
Related
Was this page helpful?