# Stream Receive (/sdks/java/how-to/queues/stream-receive)



## Overview [#overview]

A **downstream receiver** is the persistent-connection way to pull queue messages: instead of opening and tearing down a request for every batch, you open one gRPC stream and reuse it across many receive cycles. That matters for any consumer that runs continuously — a worker loop, a background processor — where reconnecting per batch would add latency and churn on both the client and the broker.

`receiveQueueMessages()` fetches a batch under manual settlement — messages come back locked, not auto-removed — so nothing leaves the queue until you explicitly settle it. Each returned `QueueMessageReceived` is settled on its own: calling `ack()` removes it from the queue immediately, while an unacknowledged message is redelivered to the next poller once the visibility window expires.

**Gotchas:** a crash between receiving and acknowledging redelivers the whole batch, so processing must be idempotent; forgetting to call `ack()` doesn't lose the message, it just delays redelivery until the timeout; and a large `pollMaxMessages` with a slow per-message handler can hold the batch past the visibility window, triggering duplicate delivery mid-processing.

## Prerequisites [#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](/sdks/java))

## Code [#code]

```java title="StreamReceiveExample.java"
package io.kubemq.example.queuesstream;

import io.kubemq.sdk.queues.*;
import java.util.UUID;

/**
 * StreamReceiveExample for Queues Stream
 */
public class StreamReceiveExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-queues-stream-receive-client";
    private static final String CHANNEL = "java-queues.stream-receive";

    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(("Message " + i).getBytes()).build());
            }

            // Poll for messages via stream
            System.out.println("Receiving messages via stream poll...\n");
            QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
                    .channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(5).build());

            // Process each message and acknowledge it
            response.getMessages().forEach(msg -> {
                System.out.println("  Received: " + new String(msg.getBody()));
                msg.ack();
            });
            System.out.println("\nReceived and acked " + response.getMessages().size() + " messages.");

            // Clean up resources
            client.deleteQueuesChannel(CHANNEL);
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }
}

```

## How It Works [#how-it-works]

* `receiveQueueMessages()` with `pollMaxMessages(10)` opens a downstream gRPC stream to the broker and returns up to 10 messages in one batch; the SDK closes the stream internally once the batch is delivered or `pollWaitTimeoutInSeconds` elapses.
* Each `QueueMessageReceived.ack()` sends a per-message acknowledgement; the broker removes acknowledged messages from the queue immediately.
* Unacknowledged messages (not acked or rejected before the visibility window expires) are redelivered to the next poller.
* Five messages are pre-sent, then all are received in a single `receiveQueueMessages()` call — this demonstrates the batch receive pattern that reduces round-trip overhead compared to polling one message at a time.

## Related [#related]

* [Pattern overview](/learn/queues/getting-started)
* [Java SDK Reference](/sdks/java/reference/queues)
* [Send & Receive](/sdks/java/tutorials/send-receive)
* [Ack All](/sdks/java/how-to/queues/ack-all)