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

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 — competing consumers, one message → one consumer.
Delivery guarantees → exactly-once — manual ack/nack, redelivery, and idempotency.
Scaling & flow → competing consumers — visibility timeout, backpressure, and adding consumers to drain a queue faster.

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

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.

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

Key Properties

Property	Behavior	Learn more
Delivery	Exactly one consumer per message	Send & Receive
Settlement	Manual `ack` / `nack` / requeue	Ack, Nack & Requeue
Visibility	Hidden from others during processing	Visibility timeout
Failure handling	Redeliver, then route to DLQ	Dead Letter Queue
Scheduling	Delay before a message becomes available	Delayed Messages
Expiration	TTL auto-discards unprocessed messages	Message expiration

Message Lifecycle

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

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
Durable history / replay	❌ Messages removed on ack	✅ Use Events Store

Need to broadcast every message to all listeners instead of distributing work? Use Events. Need a durable, replayable log? Use Events Store.

Quick Example

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()
}

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()

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();

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();

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();
}

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()

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();
}

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(())
}

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

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)

Queues are also available via the CloudEvents protocol — use any language with a CloudEvents SDK, no KubeMQ client library needed.

Queues — Point-to-Point Messaging

The concept it implements

Key Features

How It Works

Key Properties

Message Lifecycle

When to Use Queues

Quick Example

Learn More

Getting Started

Send & Receive

Ack, Nack & Requeue

Dead Letter Queue

Delayed Messages

Batch Operations

Stream API

Queue Reference

On this page