# Queues — Point-to-Point Messaging (/learn/queues)



<QueuesHero className="w-full max-h-[300px]" />

Think of Queues like a mailbox — you drop a letter in, and it waits safely until the recipient picks it up. Even if the recipient is away, the letter doesn't disappear. Each letter is delivered to exactly one person, and they confirm receipt by opening it.

KubeMQ Queues provide durable, point-to-point messaging with guaranteed delivery. Each message is persisted to a named queue and delivered to exactly one consumer, where it remains until that consumer explicitly acknowledges it. When several consumers poll the same queue, they form a pool of *competing consumers* — the broker hands each message to just one of them, load-balancing the work. Queues are ideal for task distribution, job processing, and any workflow where reliable, ordered, exactly-once processing is required.

## The concept it implements [#the-concept-it-implements]

<Callout type="info">
  Queues realize the **point-to-point** interaction style with **exactly-once processing**. To understand the underlying ideas first, see the Fundamentals layer:

  * [Interaction styles → point-to-point](/learn/concepts/interaction-styles) — competing consumers, one message → one consumer.
  * [Delivery guarantees → exactly-once](/learn/concepts/delivery-guarantees) — manual ack/nack, redelivery, and idempotency.
  * [Scaling & flow → competing consumers](/learn/concepts/scaling-and-flow) — visibility timeout, backpressure, and adding consumers to drain a queue faster.
</Callout>

## Key Features [#key-features]

* **Guaranteed delivery** — messages persist in the queue until a consumer acknowledges them. No message is lost, even if consumers restart.
* **Exactly-once processing** — each message is delivered to a single consumer with manual acknowledgment, preventing duplicate processing.
* **FIFO ordering** — messages are delivered in the order they were sent, maintaining strict first-in, first-out semantics.
* **Visibility timeout** — while a consumer processes a message, it is hidden from other consumers. If not acknowledged within the timeout, the message becomes available again.
* **Dead Letter Queue (DLQ)** — messages that exceed a configurable retry count are automatically routed to a DLQ for inspection and recovery.
* **Delayed delivery** — schedule messages to become available after a specified delay, enabling deferred processing and retry patterns.
* **Batch operations** — send and receive multiple messages in a single request for high-throughput scenarios.
* **Peek without consuming** — inspect queue contents without removing messages, useful for monitoring and debugging.
* **Message expiration (TTL)** — set time-to-live on messages so unprocessed items auto-expire.

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

A queue is a durable, first-in-first-out buffer. Producers append messages to a named queue; a pool of consumers polls the same queue and competes for messages. The broker delivers each message to **exactly one** consumer, hides it from the others while it is being processed, and removes it only after that consumer acknowledges it.

<Mermaid
  chart="graph LR
  P1[&#x22;Producer A&#x22;]
  P2[&#x22;Producer B&#x22;]
  Q{{&#x22;Queue<br/>orders&#x22;}}
  C1[&#x22;Consumer 1&#x22;]
  C2[&#x22;Consumer 2&#x22;]
  C3[&#x22;Consumer 3&#x22;]

  P1 -- send --> Q
  P2 -- send --> Q
  Q -- &#x22;deliver (1 of N)&#x22; --> C1
  Q -- &#x22;deliver (1 of N)&#x22; --> C2
  Q -- &#x22;deliver (1 of N)&#x22; --> C3
  C1 -. ack .-> Q
  C2 -. ack .-> Q
  C3 -. ack .-> Q

  class Q queue
  class P1,P2,C1,C2,C3 client"
/>

*Producers append to the queue; competing consumers each receive a different message and confirm with a dotted acknowledgment.*

## Key Properties [#key-properties]

| Property         | Behavior                                 | Learn more                                                      |
| ---------------- | ---------------------------------------- | --------------------------------------------------------------- |
| Delivery         | Exactly one consumer per message         | [Send & Receive](/learn/queues/tutorials/send-receive)          |
| Settlement       | Manual `ack` / `nack` / requeue          | [Ack, Nack & Requeue](/learn/queues/tutorials/ack-nack-requeue) |
| Visibility       | Hidden from others during processing     | [Visibility timeout](/learn/queues/how-to/visibility-timeout)   |
| Failure handling | Redeliver, then route to DLQ             | [Dead Letter Queue](/learn/queues/tutorials/dead-letter-queue)  |
| Scheduling       | Delay before a message becomes available | [Delayed Messages](/learn/queues/tutorials/delayed-messages)    |
| Expiration       | TTL auto-discards unprocessed messages   | [Message expiration](/learn/queues/how-to/message-expiration)   |

## Message Lifecycle [#message-lifecycle]

<Mermaid
  chart="stateDiagram-v2
    [*] --> Delayed: Send with delay > 0
    [*] --> Available: Send (no delay)
    Delayed --> Available: Delay expires
    Available --> Processing: Consumer receives
    Processing --> Removed: ack()
    Processing --> Available: nack() / requeue
    Processing --> Available: Visibility timeout expires
    Processing --> DeadLetter: Receive count exceeds max
    Available --> Expired: TTL reached
    Expired --> [*]: Discarded
    Removed --> [*]
    DeadLetter --> [*]: Moved to DLQ channel"
/>

*A message moves from queued to delivered to acked; on failure it is requeued or, after exhausting retries, routed to a dead letter queue.*

## When to Use Queues [#when-to-use-queues]

| Scenario                              | Queues                     | Events / Events Store                     |
| ------------------------------------- | -------------------------- | ----------------------------------------- |
| Task & job distribution               | ✅ Best choice              | ❌ No load balancing per message           |
| Exactly-once work processing          | ✅ Best choice              | ❌ At-most/at-least-once                   |
| Order-sensitive pipelines (FIFO)      | ✅ Best choice              | Use Events Store for replay               |
| Retry with backoff & dead-letter      | ✅ Best choice              | ❌ Not built in                            |
| Real-time fan-out to many subscribers | ❌ One consumer per message | ✅ Use [Events](/learn/events)             |
| Durable history / replay              | ❌ Messages removed on ack  | ✅ Use [Events Store](/learn/events-store) |

<Callout type="info">
  Need to broadcast every message to **all** listeners instead of distributing work? Use [Events](/learn/events). Need a durable, replayable log? Use [Events Store](/learn/events-store).
</Callout>

## Quick Example [#quick-example]

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="main.go"
    package main

    import (
        "context"
        "fmt"
        "log"
        "time"

        "github.com/kubemq-io/kubemq-go/v2"
    )

    func main() {
        ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
        defer cancel()

        client, err := kubemq.NewClient(ctx,
            kubemq.WithAddress("localhost", 50000),
            kubemq.WithClientId("queue-demo"),
        )
        if err != nil {
            log.Fatal(err)
        }
        defer client.Close()

        msg := kubemq.NewQueueMessage().
            SetChannel("orders").
            SetBody([]byte(`{"orderId":"ORD-1234","total":99.99}`))

        result, err := client.SendQueueMessage(ctx, msg)
        if err != nil {
            log.Fatal(err)
        }
        fmt.Printf("Sent: id=%s\n", result.MessageID)

        resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
            Channel:            "orders",
            MaxItems:           1,
            WaitTimeoutSeconds: 5,
            AutoAck:            false,
        })
        if err != nil {
            log.Fatal(err)
        }
        for _, m := range resp.Messages {
            fmt.Printf("Received: %s\n", string(m.Message.Body))
        }
        resp.AckAll()
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="main.py"
    from kubemq.queues import Client as QueuesClient
    from kubemq import QueueMessage

    client = QueuesClient(
        address="localhost:50000",
        client_id="queue-demo",
    )

    result = client.send_queue_message(
        QueueMessage(
            channel="orders",
            body=b'{"orderId":"ORD-1234","total":99.99}',
        )
    )
    print(f"Sent: id={result.id}")

    response = client.receive_queue_messages(
        channel="orders",
        max_messages=1,
        wait_timeout_in_seconds=5,
    )
    for msg in response.messages:
        print(f"Received: {msg.body.decode('utf-8')}")
        msg.ack()

    client.close()
    ```
  </Tab>

  <Tab value="Node.js">
    ```typescript title="main.ts"
    import { KubeMQClient, createQueueMessage } from 'kubemq-js';

    const client = await KubeMQClient.create({
      address: 'localhost:50000',
      clientId: 'queue-demo',
    });

    const result = await client.sendQueueMessage(
      createQueueMessage({
        channel: 'orders',
        body: JSON.stringify({ orderId: 'ORD-1234', total: 99.99 }),
      }),
    );
    console.log('Sent:', result.messageId);

    const messages = await client.receiveQueueMessages({
      channel: 'orders',
      maxMessages: 1,
      waitTimeoutSeconds: 5,
    });
    for (const msg of messages) {
      console.log('Received:', new TextDecoder().decode(msg.body));
      await msg.ack();
    }

    await client.close();
    ```
  </Tab>

  <Tab value="Java">
    ```java title="Main.java"
    QueuesClient client = QueuesClient.builder()
        .address("localhost:50000")
        .clientId("queue-demo")
        .build();

    QueueMessage msg = QueueMessage.builder()
        .channel("orders")
        .body("{\"orderId\":\"ORD-1234\",\"total\":99.99}".getBytes())
        .build();

    SendQueueMessageResult result = client.sendQueueMessage(msg);
    System.out.println("Sent: id=" + result.getMessageId());

    ReceiveQueueMessagesResponse response = client.receiveQueueMessages(
        ReceiveQueueMessagesRequest.builder()
            .channel("orders")
            .maxMessages(1)
            .waitTimeoutSeconds(5)
            .build());

    for (QueueMessageReceived m : response.getMessages()) {
        System.out.println("Received: " + new String(m.getBody()));
        m.ack();
    }

    client.close();
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="Program.cs"
    using KubeMQ.Sdk.Client;

    await using var client = new KubeMQClient(new KubeMQClientOptions());
    await client.ConnectAsync();

    var result = await client.SendQueueMessageAsync(new QueueMessage
    {
        Channel = "orders",
        Body = Encoding.UTF8.GetBytes("{\"orderId\":\"ORD-1234\",\"total\":99.99}")
    });
    Console.WriteLine($"Sent: id={result.MessageId}");

    var response = await client.ReceiveQueueMessagesAsync(new ReceiveQueueMessagesRequest
    {
        Channel = "orders",
        MaxMessages = 1,
        WaitTimeoutSeconds = 5,
    });

    foreach (var msg in response.Messages)
    {
        Console.WriteLine($"Received: {Encoding.UTF8.GetString(msg.Body.Span)}");
        await msg.AckAsync();
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="Main.kt"
    val client = QueuesClient("localhost:50000")

    val result = client.sendQueueMessage(QueueMessage(
        channel = "orders",
        body = """{"orderId":"ORD-1234","total":99.99}""".toByteArray()
    ))
    println("Sent: id=${result.messageId}")

    val response = client.receiveQueueMessages(
        channel = "orders",
        maxMessages = 1,
        waitTimeoutSeconds = 5
    )
    for (msg in response.messages) {
        println("Received: ${String(msg.body)}")
        msg.ack()
    }

    client.close()
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="main.cpp"
    #include <kubemq/client.h>
    #include <iostream>

    auto client = kubemq::QueuesClient("localhost:50000");

    kubemq::QueueMessage msg;
    msg.channel = "orders";
    msg.body = R"({"orderId":"ORD-1234","total":99.99})";

    auto result = client.sendQueueMessage(msg);
    std::cout << "Sent: id=" << result.messageId << std::endl;

    auto response = client.receiveQueueMessages("orders", 1, 5);
    for (const auto& m : response.messages) {
        std::cout << "Received: " << m.body << std::endl;
        m.ack();
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="main.rs"
    use kubemq::prelude::*;
    use kubemq::QueueMessageBuilder;

    #[tokio::main]
    async fn main() -> kubemq::Result<()> {
        let client = KubemqClient::builder()
            .host("localhost")
            .port(50000)
            .client_id("queue-demo")
            .build()
            .await?;

        let msg = QueueMessageBuilder::new()
            .channel("orders")
            .body(br#"{"orderId":"ORD-1234","total":99.99}"#.to_vec())
            .build();

        let result = client.send_queue_message(msg).await?;
        println!("Sent: id={}", result.message_id);

        // receive_queue_messages(channel, max_messages, wait_timeout_secs, auto_ack)
        let messages = client
            .receive_queue_messages("orders", 1, 5, false)
            .await?;
        for m in &messages {
            println!("Received: {}", String::from_utf8_lossy(&m.body));
            m.ack().await?;
        }

        client.close().await?;
        Ok(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="main.rb"
    require 'kubemq'

    client = KubeMQ::QueuesClient.new(address: 'localhost:50000', client_id: 'queue-demo')

    msg = KubeMQ::Queues::QueueMessage.new(
      channel: 'orders',
      body: '{"orderId":"ORD-1234","total":99.99}'
    )
    result = client.send_queue_message(msg)
    puts "Sent: id=#{result.id}"

    messages = client.receive_queue_messages(channel: 'orders', max_messages: 1, wait_timeout_seconds: 5)
    messages.each do |m|
      puts "Received: #{m.body}"
      m.ack
    end

    client.close
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="main.exs"
    {:ok, client} =
      KubeMQ.Client.start_link(address: "localhost:50000", client_id: "queue-demo")

    msg =
      KubeMQ.QueueMessage.new(
        channel: "orders",
        body: ~s({"orderId":"ORD-1234","total":99.99})
      )

    {:ok, result} = KubeMQ.Client.send_queue_message(client, msg)
    IO.puts("Sent: id=#{result.message_id}")

    {:ok, response} =
      KubeMQ.Client.receive_queue_messages(client, "orders",
        max_messages: 1,
        wait_timeout: 5_000
      )

    Enum.each(response.messages, fn m ->
      IO.puts("Received: #{m.body}")
      KubeMQ.Client.ack_queue_message(client, m)
    end)

    KubeMQ.Client.close(client)
    ```
  </Tab>
</Tabs>

<Callout type="info">
  Queues are also available via the [CloudEvents protocol](/connectors/cloudevents/how-to/queues) — use any language with a CloudEvents SDK, no KubeMQ client library needed.
</Callout>

## Learn More [#learn-more]

<Cards>
  <Card title="Getting Started" href="/learn/queues/getting-started" description="Send and receive your first queue message in 5 minutes." />

  <Card title="Send & Receive" href="/learn/queues/tutorials/send-receive" description="Complete send-receive-acknowledge cycle with metadata and tags." />

  <Card title="Ack, Nack & Requeue" href="/learn/queues/tutorials/ack-nack-requeue" description="Master the three message settlement options." />

  <Card title="Dead Letter Queue" href="/learn/queues/tutorials/dead-letter-queue" description="Route failed messages to a DLQ after max retries." />

  <Card title="Delayed Messages" href="/learn/queues/tutorials/delayed-messages" description="Schedule messages for future delivery." />

  <Card title="Batch Operations" href="/learn/queues/tutorials/batch-operations" description="Send and receive multiple messages in a single request." />

  <Card title="Stream API" href="/learn/queues/tutorials/stream-api" description="Bidirectional streaming for continuous send and receive." />

  <Card title="Queue Reference" href="/learn/queues/reference" description="Message structure, configuration, and error codes." />
</Cards>