# Queues (/integrations/spring-boot/how-to/queues)



KubeMQ queues are **durable, point-to-point** channels with **competing consumers**: every message is delivered to exactly one consumer, messages are persisted by the broker until acknowledged, and multiple instances polling the same channel share the load. Unlike Events — which the broker pushes to subscribers — the Spring adapter **polls** the broker for queue messages. The `@KubeMQQueueListener` runs one or more poll loops that pull a batch of messages each cycle, hand them to your method, and acknowledge them based on how you configure the listener.

This page covers sending, the polling listener and its tuning attributes, the acknowledgment model, advanced queue features (delay, expiration, peek, dead-letter), and the Kotlin coroutine API. For the broader queue concept, see the [core Queues guide](/learn/queues); for fire-and-forget delivery, see [Events](/integrations/spring-boot/how-to/events-and-events-store).

<Mermaid
  chart="`
flowchart LR
  P[&#x22;KubeMQTemplate<br/>sendQueueMessage&#x22;] --> Q[&#x22;KubeMQ Queue<br/>(durable, persisted)&#x22;]
  Q -->|&#x22;poll&#x22;| L1[&#x22;@KubeMQQueueListener<br/>instance A&#x22;]
  Q -->|&#x22;poll&#x22;| L2[&#x22;@KubeMQQueueListener<br/>instance B&#x22;]

  class P,L1,L2 client
  class Q queue
`"
/>

*The template enqueues durable messages; competing `@KubeMQQueueListener` instances poll the queue, and each message goes to exactly one.*

## Sending Messages [#sending-messages]

Inject `KubeMQTemplate` and call `sendQueueMessage(channel, data)`. The template serializes the payload (a `byte[]`, a `String`, or any object handled by the configured message converter) and enqueues it durably. The blocking overload also accepts a `Map<String, String>` of tags as a third argument; `sendQueueMessageAsync` returns a `CompletableFuture<Void>` for non-blocking sends.

```java title="SendReceiveRunner.java"
@Component
public class SendReceiveRunner implements ApplicationRunner {

    private final KubeMQTemplate template;

    public SendReceiveRunner(KubeMQTemplate template) {
        this.template = template;
    }

    @Override
    public void run(ApplicationArguments args) throws InterruptedException {
        for (int i = 1; i <= 3; i++) {
            template.sendQueueMessage("spring-queues.send-receive", "Queue message #" + i);
        }
    }
}
```

To send several messages to the same channel in one call, use `sendQueueMessages(channel, List<?>)`. Each element is serialized and enqueued; messages are pre-built before sending so a serialization error fails fast before anything is published.

```java title="BatchSendRunner.java"
template.sendQueueMessages("spring-queues.batch-send",
        List.of("Batch msg #1", "Batch msg #2", "Batch msg #3"));
```

<Callout type="info">
  Queue sends are durable. A message stays in the queue until a consumer acknowledges it (or it expires), so unlike Events you can send before any consumer is running and the message will be waiting when one starts polling.
</Callout>

## Consuming Messages [#consuming-messages]

Annotate a bean method with `@KubeMQQueueListener` to start a poll loop against one or more channels. The method receives a single `QueueMessageReceived`; call `getId()` for the message id and `getBody()` for the raw payload bytes.

```java title="SendReceiveListener.java"
package io.kubemq.spring.boot.examples.queues;

import io.kubemq.sdk.queues.QueueMessageReceived;
import io.kubemq.spring.boot.autoconfigure.listener.KubeMQQueueListener;
import java.nio.charset.StandardCharsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class SendReceiveListener {

    private static final Logger log = LoggerFactory.getLogger(SendReceiveListener.class);

    @KubeMQQueueListener(channels = "spring-queues.send-receive", autoAck = "true")
    public void onMessage(QueueMessageReceived msg) {
        byte[] body = msg.getBody();
        String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
        log.info("Received queue message: id={} body={}", msg.getId(), text);
    }
}
```

To process messages in batches, set `batch = "true"` and declare the parameter as `List<QueueMessageReceived>`. The listener delivers up to `maxPollMessages` messages per cycle in a single invocation:

```java title="BatchSendListener.java"
@Component
public class BatchSendListener {

    private static final Logger log = LoggerFactory.getLogger(BatchSendListener.class);

    @KubeMQQueueListener(channels = "spring-queues.batch-send", batch = "true",
            maxPollMessages = "10", autoAck = "true")
    public void onMessages(List<QueueMessageReceived> messages) {
        log.info("Received batch of {} messages", messages.size());
        for (QueueMessageReceived msg : messages) {
            String body = msg.getBody() != null
                    ? new String(msg.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("  Batch message: id={} body={}", msg.getId(), body);
        }
    }
}
```

The `channels` attribute supports SpEL expressions and property placeholders, so you can drive the channel name from configuration — for example `channels = "${kubemq.channels.orders}"`.

## Listener Tuning [#listener-tuning]

Each `@KubeMQQueueListener` attribute is a `String` so it can carry a literal, a SpEL expression, or a `${...}` placeholder. When an attribute is left blank, the listener falls back to the matching default under `kubemq.listener.queues.*`. The attributes and their defaults:

<TypeTable
  type="{
  pollTimeout: {
    description: 'Maximum time the broker waits for messages before returning an empty poll.',
    type: 'duration',
    default: '5s (kubemq.listener.queues.poll-timeout)',
  },
  maxPollMessages: {
    description: 'Maximum number of messages returned per poll cycle.',
    type: 'int',
    default: '1 (kubemq.listener.queues.max-poll-messages)',
  },
  visibilityTimeout: {
    description: 'How long a received message stays hidden from other consumers before becoming visible again if not acknowledged.',
    type: 'duration',
    default: '30s (kubemq.listener.queues.visibility-timeout)',
  },
  autoAck: {
    description: 'Whether messages are acknowledged automatically on delivery.',
    type: 'boolean',
    default: 'false (kubemq.listener.queues.auto-ack)',
  },
  concurrency: {
    description: 'Number of concurrent poll loops for this listener.',
    type: 'int',
    default: '1 (kubemq.listener.concurrency)',
  },
}"
/>

Set the global defaults under the `kubemq.listener.queues` prefix so every queue listener inherits them, then override per listener where needed:

```yaml title="application.yml"
kubemq:
  listener:
    concurrency: 1
    queues:
      poll-timeout: 5s
      max-poll-messages: 1
      visibility-timeout: 30s
      auto-ack: false
```

For example, a high-throughput listener that pulls ten messages per poll across four concurrent loops and acknowledges manually:

```java title="OrderListener.java"
@KubeMQQueueListener(
        channels = "spring-queues.orders",
        maxPollMessages = "10",
        concurrency = "4",
        visibilityTimeout = "60",
        autoAck = "false")
public void onMessages(List<QueueMessageReceived> messages) {
    // process each message and ack/reject individually
}
```

<Callout type="info">
  `visibilityTimeout` and `pollTimeout` are expressed in **seconds** when given as a bare number on the annotation (for example `visibilityTimeout = "60"`), matching the underlying KubeMQ Java SDK poll request. The `kubemq.listener.queues.*` properties accept Spring `Duration` syntax such as `5s` or `30s`.
</Callout>

## Acknowledgment Model [#acknowledgment-model]

With `autoAck = "false"` (the default), your method owns the outcome of every message:

* **`msg.ack()`** — the message was processed successfully; remove it from the queue.
* **`msg.reject()`** — processing failed; return the message to the queue so it can be redelivered (and eventually dead-lettered, if configured).

The ack/reject example acknowledges valid messages and rejects ones it cannot process:

```java title="AckRejectListener.java"
@KubeMQQueueListener(channels = "spring-queues.ack-reject", autoAck = "false")
public void onMessage(QueueMessageReceived msg) {
    byte[] body = msg.getBody();
    String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
    if ("INVALID".equals(text)) {
        log.warn("Rejecting invalid message: id={}", msg.getId());
        msg.reject();
    } else {
        log.info("Processing and acknowledging: id={} body={}", msg.getId(), text);
        msg.ack();
    }
}
```

When you prefer to drain a queue imperatively rather than through the annotation, inject the `QueuesClient` directly, build a `QueuesPollRequest` with `autoAckMessages(false)`, and acknowledge each returned message after processing. This "ack-all" pattern pulls a whole batch, processes it, then acks every message:

```java title="AckAllRunner.java"
QueuesPollRequest request = QueuesPollRequest.builder()
        .channel("spring-queues.ack-all")
        .pollMaxMessages(10)
        .pollWaitTimeoutInSeconds(5)
        .autoAckMessages(false)
        .build();

QueuesPollResponse response = queuesClient.receiveQueuesMessages(request);
if (response.getMessages() != null) {
    for (QueueMessageReceived msg : response.getMessages()) {
        // process ...
        msg.ack();
    }
    log.info("Acknowledged all {} messages", response.getMessages().size());
}
```

## Advanced Queue Features [#advanced-queue-features]

The fluent builder (`template.newQueueMessage(...)`, covered below) and the `QueuesClient` expose queue-specific behaviors beyond plain send and receive.

<Accordions>
  <Accordion title="Delayed delivery">
    `withDelay(Duration)` holds a message invisible to consumers for a fixed period after it is enqueued. The listener only sees it once the delay elapses.

    ```java title="DelayedRunner.java"
    template.newQueueMessage("Delayed message")
            .toChannel("spring-queues.delayed")
            .withDelay(Duration.ofSeconds(5))
            .send();
    ```
  </Accordion>

  <Accordion title="Message expiration">
    `withExpiration(Duration)` sets a time-to-live. If the message is not consumed within the window, the broker drops it instead of delivering it.

    ```java title="ExpirationRunner.java"
    template.newQueueMessage("Expiring message")
            .toChannel("spring-queues.expiration")
            .withExpiration(Duration.ofSeconds(10))
            .send();
    ```
  </Accordion>

  <Accordion title="Peek (non-consuming receive)">
    To inspect a message without consuming it, poll with `autoAckMessages(false)` and then `reject()` it — the message returns to the queue for later consumption.

    ```java title="PeekRunner.java"
    QueuesPollRequest peekRequest = QueuesPollRequest.builder()
            .channel("spring-queues.peek")
            .pollMaxMessages(1)
            .pollWaitTimeoutInSeconds(5)
            .autoAckMessages(false)
            .build();

    QueuesPollResponse response = queuesClient.receiveQueuesMessages(peekRequest);
    if (response.getMessages() != null && !response.getMessages().isEmpty()) {
        QueueMessageReceived msg = response.getMessages().get(0);
        // inspect msg.getBody() ...
        msg.reject(); // returned to the queue
    }
    ```
  </Accordion>

  <Accordion title="Dead-letter routing">
    `withDeadLetterQueue(channel, maxReceiveCount)` routes a message to a dead-letter channel after it has been rejected the given number of times. Below, the message is rejected three times on `spring-queues.dead-letter` and then moves to `spring-queues.dead-letter.dlq`, where a second listener consumes it.

    ```java title="DeadLetterRunner.java"
    template.newQueueMessage("DLQ test message")
            .toChannel("spring-queues.dead-letter")
            .withDeadLetterQueue("spring-queues.dead-letter.dlq", 3)
            .send();
    ```

    ```java title="DeadLetterListener.java"
    @Component
    public class DeadLetterListener {

        @KubeMQQueueListener(channels = "spring-queues.dead-letter", autoAck = "false")
        public void onMessage(QueueMessageReceived msg) {
            // reject() repeatedly; after 3 attempts the broker routes to the DLQ
            msg.reject();
        }

        @KubeMQQueueListener(channels = "spring-queues.dead-letter.dlq", autoAck = "true")
        public void onDeadLetter(QueueMessageReceived msg) {
            byte[] body = msg.getBody();
            String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
            log.info("Dead-letter received: body={}", text);
        }
    }
    ```
  </Accordion>
</Accordions>

## Error Backoff [#error-backoff]

When a poll loop throws — for example because the broker is unreachable — the listener backs off before retrying instead of spinning. The backoff starts at `error-backoff-initial`, multiplies by `error-backoff-multiplier` after each failure, and is capped at `error-backoff-max`:

```yaml title="application.yml"
kubemq:
  listener:
    queues:
      error-backoff-initial: 1s
      error-backoff-max: 30s
      error-backoff-multiplier: 2.0
```

With these defaults the retry delay grows 1s → 2s → 4s → 8s → 16s → 30s (capped) and resets once a poll succeeds.

## Fluent Builder [#fluent-builder]

`template.newQueueMessage(data)` returns a `KubeMQQueueMessageBuilder` for composing a message with channel, tags, metadata, and the advanced options above in one chain, then `send()` (blocking) or `sendAsync()` (returning a `CompletableFuture<Void>`):

```java title="FluentBuildersRunner.java"
template.newQueueMessage("Delayed queue msg")
        .toChannel("spring-fluent.queues")
        .withDelay(Duration.ofSeconds(1))
        .withExpiration(Duration.ofSeconds(30))
        .send();
```

The builder also exposes `withTag(key, value)`, `withTags(Map)`, `withMetadata(String)`, and `withDeadLetterQueue(channel, maxReceiveCount)`.

## Java and Kotlin [#java-and-kotlin]

The starter is fully usable from Java. The optional `kubemq-spring-boot-starter-kotlin` module adds coroutine-friendly `suspend` extensions on `KubeMQTemplate`; `sendQueueMessageSuspend` delegates to `sendQueueMessageAsync` and suspends until the send completes, honoring structured-concurrency cancellation.

<Tabs groupId="spring-lang" items="['Java', 'Kotlin']">
  <Tab value="Java">
    ```java title="OrderProducer.java"
    import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate;
    import org.springframework.stereotype.Service;

    @Service
    public class OrderProducer {

        private final KubeMQTemplate template;

        public OrderProducer(KubeMQTemplate template) {
            this.template = template;
        }

        public void enqueue(String order) {
            template.sendQueueMessage("spring-queues.orders", order);
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="OrderProducer.kt"
    import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate
    import io.kubemq.spring.boot.kotlin.sendQueueMessageSuspend
    import org.springframework.stereotype.Service

    @Service
    class OrderProducer(private val template: KubeMQTemplate) {

        suspend fun enqueue(order: String) {
            template.sendQueueMessageSuspend("spring-queues.orders", order)
        }
    }
    ```

    `sendQueueMessageSuspend` also has an overload that takes a `Map<String, String>` of tags as a third argument.
  </Tab>
</Tabs>

## Running the Examples [#running-the-examples]

The queue listeners and producers above need a running KubeMQ broker. The Spring Boot starter connects over gRPC on port `50000`, which is also the starter's default address (`localhost:50000`):

<RunKubeMQ ports="[50000, 9090]" />

Port `50000` is the native gRPC port the starter uses; port `9090` is the broker's shared HTTP/REST and dashboard endpoint, exposed here for convenience. The Spring Boot integration does not require the HTTP connector — it speaks gRPC directly through the KubeMQ Java SDK.

<Callout type="warn">
  **Visibility timeout vs. auto-ack — choose your delivery guarantee.** With `autoAck = "true"` a message is acknowledged the moment it is delivered, so a crash mid-processing loses it (at-most-once). With `autoAck = "false"` the message stays hidden only for `visibilityTimeout`; if your method does not `ack()` within that window — because it is still working or it crashed — the message becomes visible again and is redelivered to another consumer (at-least-once). Set `visibilityTimeout` comfortably above your worst-case processing time, and make handlers idempotent so a redelivery is safe.
</Callout>

## Next Steps [#next-steps]

<Cards>
  <Card title="Events & Events Store" href="/integrations/spring-boot/how-to/events-and-events-store" description="Fire-and-forget events and durable, replayable events store messages." />

  <Card title="Commands & Queries" href="/integrations/spring-boot/how-to/commands-and-queries" description="Synchronous request-response with the template and handler annotations." />

  <Card title="Reference" href="/integrations/spring-boot/reference/configuration" description="Complete kubemq.* configuration properties and the KubeMQTemplate API." />
</Cards>