# Queues (/connectors/mqtt/how-to/queues) Queues are durable, **point-to-point** work queues over the MQTT connector. A message is enqueued in KubeMQ and delivered to **exactly one** consumer (competing-consumer semantics). Over MQTT the produce and consume sides are asymmetric: producing is a plain publish, while consuming requires an MQTT 5.0 **shared subscription**. ## Overview [#overview] The `queues/` prefix selects the Queues pattern (`/` → `.` in the channel: `queues/jobs/email` → `jobs.email`). A producer plain-publishes to `queues/` on any MQTT version; a consumer subscribes to `$share//queues/` on **MQTT 5.0 only**. Acknowledgement is **ack-on-PUBACK**: the consumer's `PUBACK` removes the message from the queue. | Operation | MQTT action | KubeMQ mapping | | --------- | --------------------------------------------------------------- | ------------------------------------------------- | | Produce | `PUBLISH queues/` (any QoS, any version) | `SendQueueMessage` — durably enqueued | | Consume | `SUBSCRIBE $share//queues/` (QoS ≥ 1, MQTT 5.0) | Credit-driven `Get`; each message to one consumer | | Ack | `PUBACK` from the consumer | Message acknowledged and removed | | Redeliver | no PUBACK within `QueueAckTimeoutSeconds` (30 s), or disconnect | NAck → redelivered to another consumer | **Queues are publish-only over MQTT; consuming requires MQTT 5.0.** You produce with a plain `PUBLISH queues/`, but you can only consume through an MQTT 5.0 **shared subscription** `$share//queues/`. A plain `queues/` subscribe is rejected with SUBACK `0x83`, and a QoS-0 shared subscribe is rejected with SUBACK `0x83`. **MQTT 3.1.1 has no shared subscriptions, so it cannot consume Queues at all** — a 3.1.1 client can produce but never consume. The Ruby `mqtt` gem is 3.1.1-only and is therefore produce-only for Queues. ## How it works [#how-it-works] A producer enqueues messages on the plain `queues/` topic. Competing consumers attach via `$share//queues/`; the broker hands each message to exactly one of them, and the consumer's `PUBACK` acks it. *Each queued message is delivered to exactly one shared-subscription consumer; the consumer's `PUBACK` acknowledges and removes it.* **The `$share` group name is audit-only.** Unlike standard MQTT shared subscriptions, the `` token in `$share//queues/` does **not** create per-group copies. All consumers — across every group name — compete in one KubeMQ queue pool; a message goes to exactly one of them regardless of group. The group is recorded only in connection metadata and metrics. For per-group fan-out, use [Events](/connectors/mqtt/how-to/events) or [Events Store](/connectors/mqtt/how-to/events-store). ## Produce and consume [#produce-and-consume] Each example consumes through an MQTT 5.0 shared subscription `$share/g1/queues/demo/q`, then produces a plain publish to `queues/demo/q`, and confirms the message is delivered to exactly one consumer (the auto-`PUBACK` acks it). Every client reads the broker endpoint from `KUBEMQ_MQTT_URL` (default `tcp://localhost:1883`). Ruby is omitted from the consume round-trip — the `mqtt` gem is MQTT 3.1.1-only and shared subscriptions require MQTT 5.0; the produce side ([Events](/connectors/mqtt/how-to/events) shows the same plain-publish form) works from Ruby. ```go package main import ( "context" "fmt" "log" "net" "os" "strings" "time" "github.com/eclipse/paho.golang/paho" "github.com/google/uuid" ) const produceTopic = "queues/demo/q" // plain publish; channel demo.q const consumeTopic = "$share/g1/queues/demo/q" // MQTT 5.0 shared subscription func brokerURL() string { if u := os.Getenv("KUBEMQ_MQTT_URL"); u != "" { return u } return "tcp://localhost:1883" } func tcpAddr(raw string) string { for _, pfx := range []string{"tcp://", "ws://", "tls://"} { if strings.HasPrefix(raw, pfx) { return raw[len(pfx):] } } return raw } func dial(ctx context.Context, addr, id string, onMsg func(paho.PublishReceived) (bool, error)) *paho.Client { conn, err := net.Dial("tcp", addr) if err != nil { log.Fatalf("dial: %v", err) } cfg := paho.ClientConfig{Conn: conn} if onMsg != nil { cfg.OnPublishReceived = []func(paho.PublishReceived) (bool, error){onMsg} } c := paho.NewClient(cfg) ack, err := c.Connect(ctx, &paho.Connect{ClientID: id, KeepAlive: 30, CleanStart: true}) if err != nil { log.Fatalf("connect: %v", err) } if ack.ReasonCode != 0 { log.Fatalf("CONNACK reason=0x%02X", ack.ReasonCode) } return c } func main() { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() addr := tcpAddr(brokerURL()) sfx := uuid.NewString()[:8] // 1. CONSUME — MQTT 5.0 shared subscription. Returning true sends PUBACK, // which is how KubeMQ marks the message acknowledged. got := make(chan string, 1) consumer := dial(ctx, addr, "go-q-worker-"+sfx, func(pr paho.PublishReceived) (bool, error) { got <- string(pr.Packet.Payload) return true, nil }) subAck, err := consumer.Subscribe(ctx, &paho.Subscribe{ Subscriptions: []paho.SubscribeOptions{{Topic: consumeTopic, QoS: 1}}, }) if err != nil { log.Fatalf("subscribe: %v", err) } // SUBACK 0x83 = plain queues/ subscribe, QoS-0 shared subscribe, or wrong pattern. if subAck.Reasons[0] == 0x83 { log.Fatal("SUBACK 0x83: use the $share//queues/ form at QoS >= 1") } time.Sleep(300 * time.Millisecond) // 2. PRODUCE — plain publish to queues/. producer := dial(ctx, addr, "go-q-prod-"+sfx, nil) pubAck, err := producer.Publish(ctx, &paho.Publish{ Topic: produceTopic, QoS: 1, Payload: []byte(`{"job":1}`), }) if err != nil { log.Fatalf("publish: %v", err) } if pubAck.ReasonCode != 0 { log.Fatalf("PUBACK reason=0x%02X", pubAck.ReasonCode) } // 3. RECEIVE. select { case msg := <-got: fmt.Printf("consumed: %s (acked via PUBACK)\n", msg) case <-ctx.Done(): log.Fatal("timed out waiting for the queue message") } _ = consumer.Disconnect(&paho.Disconnect{ReasonCode: 0}) _ = producer.Disconnect(&paho.Disconnect{ReasonCode: 0}) } ``` ```python import os import threading import time import paho.mqtt.client as mqtt from paho.mqtt.enums import CallbackAPIVersion PRODUCE_TOPIC = "queues/demo/q" # plain publish; channel demo.q CONSUME_TOPIC = "$share/g1/queues/demo/q" # MQTT 5.0 shared subscription def parse_url(url: str) -> tuple[str, int]: scheme, rest = url.split("://", 1) host, _, port = rest.rstrip("/").partition(":") return host, int(port) if port else {"tcp": 1883, "tls": 8883, "ws": 8083}[scheme] def main() -> None: host, port = parse_url(os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883")) received = threading.Event() payload: list[bytes] = [] subscribed = threading.Event() # 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 -> SUBACK 0x83). consumer = mqtt.Client(CallbackAPIVersion.VERSION2, client_id="py-q-consumer", protocol=mqtt.MQTTv5) consumer.on_connect = lambda c, *_: c.subscribe(CONSUME_TOPIC, qos=1) consumer.on_subscribe = lambda *_: subscribed.set() consumer.on_message = lambda c, u, m: (payload.append(m.payload), received.set()) consumer.connect(host, port, keepalive=30) consumer.loop_start() if not subscribed.wait(timeout=10): raise TimeoutError("shared subscription not established") # 2. PRODUCE — plain publish to queues/. producer = mqtt.Client(CallbackAPIVersion.VERSION2, client_id="py-q-producer", protocol=mqtt.MQTTv5) producer.connect(host, port, keepalive=30) producer.loop_start() producer.publish(PRODUCE_TOPIC, payload=b'{"job": 1}', qos=1).wait_for_publish(10) # 3. RECEIVE. The auto-PUBACK acks the message to KubeMQ. if not received.wait(timeout=15): raise TimeoutError("timed out waiting for the queue message") print(f"consumed: {payload[0].decode()!r} (acked via PUBACK)") producer.loop_stop(); producer.disconnect() consumer.loop_stop(); consumer.disconnect() if __name__ == "__main__": main() ``` ```java import java.nio.charset.StandardCharsets; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.eclipse.paho.mqttv5.client.MqttAsyncClient; import org.eclipse.paho.mqttv5.client.MqttCallback; import org.eclipse.paho.mqttv5.client.MqttConnectionOptions; import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse; import org.eclipse.paho.mqttv5.client.IMqttToken; import org.eclipse.paho.mqttv5.common.MqttException; import org.eclipse.paho.mqttv5.common.MqttMessage; import org.eclipse.paho.mqttv5.common.packet.MqttProperties; public final class Main { private static final String PRODUCE_TOPIC = "queues/demo/q"; // plain publish; channel demo.q private static final String CONSUME_TOPIC = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription public static void main(String[] args) throws Exception { String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883"); MqttConnectionOptions opts = new MqttConnectionOptions(); opts.setCleanStart(true); opts.setKeepAliveInterval(30); CountDownLatch received = new CountDownLatch(1); String[] body = {null}; // 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 -> SUBACK 0x83). // Paho auto-sends PUBACK when messageArrived returns, acking the message. MqttAsyncClient consumer = new MqttAsyncClient(broker, "java-q-consumer-" + UUID.randomUUID().toString().substring(0, 6)); consumer.setCallback(new MqttCallback() { public void messageArrived(String t, MqttMessage m) { body[0] = new String(m.getPayload(), StandardCharsets.UTF_8); received.countDown(); } public void disconnected(MqttDisconnectResponse r) {} public void mqttErrorOccurred(MqttException e) {} public void deliveryComplete(IMqttToken t) {} public void connectComplete(boolean reconnect, String uri) {} public void authPacketArrived(int code, MqttProperties props) {} }); consumer.connect(opts).waitForCompletion(10_000); IMqttToken subToken = consumer.subscribe(CONSUME_TOPIC, 1); subToken.waitForCompletion(10_000); // SUBACK 0x83 = plain queues/ subscribe, QoS-0 shared subscribe, or wrong pattern. if (subToken.getGrantedQos()[0] == 0x83) { throw new IllegalStateException("SUBACK 0x83: use $share//queues/ at QoS >= 1"); } Thread.sleep(400); // 2. PRODUCE — plain publish to queues/. MqttAsyncClient producer = new MqttAsyncClient(broker, "java-q-producer-" + UUID.randomUUID().toString().substring(0, 6)); producer.connect(opts).waitForCompletion(10_000); MqttMessage msg = new MqttMessage("{\"job\":1}".getBytes(StandardCharsets.UTF_8)); msg.setQos(1); producer.publish(PRODUCE_TOPIC, msg).waitForCompletion(5_000); // 3. RECEIVE. if (!received.await(15, TimeUnit.SECONDS)) { throw new IllegalStateException("timed out waiting for the queue message"); } System.out.printf("consumed: %s (acked via PUBACK)%n", body[0]); producer.disconnect().waitForCompletion(5_000); producer.close(); consumer.disconnect().waitForCompletion(5_000); consumer.close(); } } ``` ```typescript import mqtt, { type MqttClient } from "mqtt"; const PRODUCE_TOPIC = "queues/demo/q"; // plain publish; channel demo.q const CONSUME_TOPIC = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription function connect(url: string, clientId: string): Promise { return new Promise((resolve, reject) => { const client = mqtt.connect(url, { clientId, protocolVersion: 5, clean: true, keepalive: 30 }); client.once("connect", () => resolve(client)); client.once("error", reject); }); } async function main(): Promise { const url = process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883"; // 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 / plain queues/ -> SUBACK 0x83). const consumer = await connect(url, "js-q-consumer"); const received = new Promise((resolve) => { consumer.on("message", (_topic, payload) => resolve(payload.toString())); }); await new Promise((resolve, reject) => { consumer.subscribe(CONSUME_TOPIC, { qos: 1 }, (err) => err ? reject(new Error("subscribe rejected — use $share//queues/ at QoS >= 1")) : resolve(), ); }); await new Promise((r) => setTimeout(r, 300)); // 2. PRODUCE — plain publish to queues/. const producer = await connect(url, "js-q-producer"); await new Promise((resolve, reject) => { producer.publish(PRODUCE_TOPIC, JSON.stringify({ job: 1 }), { qos: 1 }, (err) => err ? reject(err) : resolve(), ); }); // 3. RECEIVE. The auto-PUBACK acks the message to KubeMQ. console.log(`consumed: ${await received} (acked via PUBACK)`); await producer.endAsync(); await consumer.endAsync(); } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp using System.Text; using MQTTnet; using MQTTnet.Client; using MQTTnet.Protocol; static (string host, int port) Endpoint() { var url = Environment.GetEnvironmentVariable("KUBEMQ_MQTT_URL") ?? "tcp://localhost:1883"; foreach (var pfx in new[] { "tcp://", "tls://", "ws://" }) if (url.StartsWith(pfx, StringComparison.OrdinalIgnoreCase)) url = url[pfx.Length..]; var parts = url.TrimEnd('/').Split(':'); return (parts[0], parts.Length > 1 ? int.Parse(parts[1]) : 1883); } var (host, port) = Endpoint(); const string produceTopic = "queues/demo/q"; // plain publish; channel demo.q const string consumeTopic = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription var factory = new MqttFactory(); var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); MqttClientOptions Options(string id) => new MqttClientOptionsBuilder() .WithTcpServer(host, port) .WithClientId($"{id}-{Guid.NewGuid():N}"[..26]) .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500) .WithCleanSession(true) .Build(); // 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 / plain queues/ -> SUBACK 0x83). using var consumer = factory.CreateMqttClient(); consumer.ApplicationMessageReceivedAsync += e => { received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment)); return Task.CompletedTask; }; await consumer.ConnectAsync(Options("csharp-q-consumer")); var subResult = await consumer.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(f => f.WithTopic(consumeTopic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)) .Build()); if (subResult.Items.First().ResultCode > MqttClientSubscribeResultCode.GrantedQoS2) throw new Exception("subscribe rejected — use $share//queues/ at QoS >= 1"); await Task.Delay(300); // 2. PRODUCE — plain publish to queues/. using var producer = factory.CreateMqttClient(); await producer.ConnectAsync(Options("csharp-q-producer")); var pubResult = await producer.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(produceTopic) .WithPayload(Encoding.UTF8.GetBytes("{\"job\":1}")) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build()); if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success) throw new Exception($"PUBACK reason: {pubResult.ReasonCode}"); // 3. RECEIVE. The auto-PUBACK acks the message to KubeMQ. using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); Console.WriteLine($"consumed: {await received.Task.WaitAsync(cts.Token)} (acked via PUBACK)"); await producer.DisconnectAsync(); await consumer.DisconnectAsync(); ``` ```rust use rumqttc::v5::mqttbytes::v5::Packet; use rumqttc::v5::mqttbytes::QoS; use rumqttc::v5::{AsyncClient, Event, MqttOptions}; use std::env; use tokio::sync::oneshot; use tokio::time::{timeout, Duration}; use uuid::Uuid; const PRODUCE_TOPIC: &str = "queues/demo/q"; // plain publish; channel demo.q const CONSUME_TOPIC: &str = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription fn parse_host_port(url: &str) -> (String, u16) { let stripped = url .trim_start_matches("tcp://") .trim_start_matches("tls://") .trim_start_matches("ws://"); let host_port = stripped.split('/').next().unwrap_or(stripped); let mut parts = host_port.splitn(2, ':'); let host = parts.next().unwrap_or("localhost").to_string(); let port: u16 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(1883); (host, port) } #[tokio::main] async fn main() -> Result<(), Box> { let url = env::var("KUBEMQ_MQTT_URL").unwrap_or_else(|_| "tcp://localhost:1883".to_string()); let (host, port) = parse_host_port(&url); let sfx = &Uuid::new_v4().to_string()[..8]; // 1. CONSUME — MQTT 5.0 shared subscription. rumqttc auto-sends PUBACK for // QoS 1, which acks the message and removes it from the queue. let mut con_opts = MqttOptions::new(format!("rust-q-con-{sfx}"), &host, port); con_opts.set_keep_alive(Duration::from_secs(30)); let (consumer, mut con_loop) = AsyncClient::new(con_opts, 10); let (msg_tx, msg_rx) = oneshot::channel::(); tokio::spawn(async move { let mut msg = Some(msg_tx); loop { match con_loop.poll().await { Ok(Event::Incoming(Packet::Publish(p))) => { if let Some(tx) = msg.take() { let _ = tx.send(String::from_utf8_lossy(&p.payload).to_string()); } return; } Ok(_) => {} Err(e) => { eprintln!("event loop: {e}"); return; } } } }); consumer.subscribe(CONSUME_TOPIC, QoS::AtLeastOnce).await?; tokio::time::sleep(Duration::from_millis(400)).await; // SUBACK round-trip // 2. PRODUCE — plain publish to queues/. let mut prod_opts = MqttOptions::new(format!("rust-q-prod-{sfx}"), &host, port); prod_opts.set_keep_alive(Duration::from_secs(30)); let (producer, mut prod_loop) = AsyncClient::new(prod_opts, 10); tokio::spawn(async move { while prod_loop.poll().await.is_ok() {} }); producer.publish(PRODUCE_TOPIC, QoS::AtLeastOnce, false, br#"{"job":1}"#.as_ref()).await?; // 3. RECEIVE. let body = timeout(Duration::from_secs(15), msg_rx).await??; println!("consumed: {body} (acked via PUBACK)"); Ok(()) } ``` ## Ack model and redelivery [#ack-model-and-redelivery] The connector acks on `PUBACK`: | Event | Outcome | | ---------------------------------------------------------------------- | --------------------------------------------------- | | Consumer sends `PUBACK` within `QueueAckTimeoutSeconds` (default 30 s) | Message acknowledged — removed from the queue | | No `PUBACK` within the timeout | NAck → message redelivered to another consumer | | Consumer disconnects with an unacked message | Immediate NAck → message requeued at once (no loss) | Because a crash before `PUBACK` requeues the message, **consumers must be idempotent** — a message can arrive more than once. Use `clean_session=false` (3.1.1) or `session_expiry_interval > 0` (5.0) so a brief reconnect restores the subscription without re-subscribing. ## Subscribe rejections [#subscribe-rejections] | SUBACK reason | Cause | | ------------- | -------------------------------------------------------------------------------------------------------- | | `0x83` | Plain `queues/` subscribe (no `$share`), QoS-0 shared subscribe, or `$share` on a non-Queues pattern | | `0xA2` | A wildcard inside a queues filter | ## Related [#related]