# RabbitMQ (AMQP 0-9-1) (/connectors/rabbitmq)



Point a RabbitMQ app at KubeMQ by changing only the connection string. The **RabbitMQ
(AMQP 0-9-1) connector** is a built-in, wire-protocol bridge inside kubemq-server that
speaks the RabbitMQ wire dialect natively — any standard AMQP 0-9-1 client (`amqp091-go`,
`pika`, `amqp-client`, `amqplib`, `RabbitMQ.Client`, `bunny`, `lapin`) talks to KubeMQ's
Queues with no KubeMQ SDK, no library swap, and no code rewrite.

## What is the RabbitMQ connector [#what-is-the-rabbitmq-connector]

[AMQP 0-9-1](https://www.rabbitmq.com/tutorials/amqp-concepts) is the wire protocol that
RabbitMQ popularized: a client opens a connection, multiplexes *channels* over it, declares
*queues* and *exchanges*, and publishes messages routed by exchange type and routing key.
The KubeMQ connector accepts every one of these operations from a stock client and bridges
them onto KubeMQ — it is a *gateway*, not a client library, so your application only needs
its existing AMQP 0-9-1 client.

<Callout type="info">
  **Mental model — everything is a Queue.** Every AMQP queue maps to exactly one KubeMQ
  **Queue** channel named `amqp.{vhost}.{queue}`. Exchanges and bindings are **virtual**,
  connector-side routing metadata resolved at publish time — not data stores. AMQP only ever
  touches the KubeMQ Queue primitive, so the connector's "patterns" mirror AMQP routing
  concepts (work queues, pub/sub, routing, topics, RPC), not KubeMQ's five messaging
  patterns.
</Callout>

Key capabilities:

* **Drop-in connection-string migration** — keep your RabbitMQ client and code; change only
  the broker host in the URL.
* **Everything is a Queue** — every AMQP queue is a durable KubeMQ Queue channel
  `amqp.{vhost}.{queue}`; exchanges (default, direct, fanout, topic, headers) route to those
  queues virtually at publish time.
* **Native RPC** — request/reply uses RabbitMQ's `amq.rabbitmq.reply-to` (direct reply-to);
  there is no gRPC responder.
* **Cross-protocol interop** — a message published over AMQP to `amqp.default.orders` is
  consumable by a gRPC or REST KubeMQ client on the same channel, and vice-versa.

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

An AMQP client publishes to an exchange with a routing key. The connector resolves the
exchange routing (default / direct / fanout / topic / headers) to a set of target queues
*at publish time*, then writes each message to that queue's KubeMQ Queue channel through the
message broker. A consumer on the same queue — over AMQP or any other KubeMQ transport —
receives it.

<Mermaid
  chart="`
graph LR
APP[&#x22;AMQP 0-9-1 client<br/>(any language)&#x22;]
CONN[&#x22;RabbitMQ connector<br/>:5672 / :5671&#x22;]
ROUTE{{&#x22;Exchange routing<br/>(virtual, at publish time)&#x22;}}
BROKER[&#x22;Message Broker&#x22;]
SUB[&#x22;Consumer<br/>(AMQP / gRPC / REST)&#x22;]

APP -- &#x22;basic.publish(exchange, key)&#x22; --> CONN
CONN -- &#x22;resolve bindings&#x22; --> ROUTE
ROUTE -- &#x22;amqp.{vhost}.{queue}&#x22; --> BROKER
BROKER -. &#x22;basic.deliver&#x22; .-> SUB

class APP,SUB client
class CONN connector
class ROUTE,BROKER broker
`"
/>

*A publish to queue `orders` on vhost `/` resolves to the KubeMQ Queue channel `amqp.default.orders`; any consumer on that channel — AMQP or gRPC/REST — receives the message.*

## Ports & protocol surface [#ports--protocol-surface]

| Port   | Transport              | Protocol                           | Notes                                                                                                |
| ------ | ---------------------- | ---------------------------------- | ---------------------------------------------------------------------------------------------------- |
| `5672` | Plain TCP (SASL PLAIN) | AMQP 0-9-1 (RabbitMQ wire dialect) | Default plain listener. **Shared with the AMQP 1.0 connector** via the internal `amqpmux`.           |
| `5671` | TLS / AMQPS over TCP   | AMQP 0-9-1                         | Binds only when the server-global `Security` block is configured. Shared TLS listener with AMQP 1.0. |

A single `amqpmux` listener accepts every connection on `5672`/`5671`, reads the 8-byte AMQP
protocol header, and routes it to the matching dialect engine — so AMQP 0-9-1 and AMQP 1.0
coexist on the same ports. SASL is **PLAIN only**. The AMQP vhost `/` maps to the
connector's configured `DefaultVhost` segment (literal `"default"`); see
[Architecture](/connectors/rabbitmq/concepts/architecture) for the dispatch detail.

## Publish to a queue [#publish-to-a-queue]

The example below declares a queue, publishes one `text/plain` message to it through the
default exchange (routing key = queue name), and consumes it back. Queue `hello` on vhost
`/` lands on the KubeMQ Queue channel `amqp.default.hello`. Every client reads the broker
endpoint from `KUBEMQ_AMQP_URL` (default `amqp://guest:guest@localhost:5672/`).

<Callout type="info" title="Which to use">
  This is the same round trip as the [Getting started](/connectors/rabbitmq/tutorials/getting-started)
  tutorial, shown here inline for reference. For a step-by-step walkthrough — enabling the
  connector, running a local broker, and verifying each step — use Getting started instead.
</Callout>

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"log"
    	"os"
    	"time"

    	amqp "github.com/rabbitmq/amqp091-go"
    )

    func amqpURL() string {
    	if v := os.Getenv("KUBEMQ_AMQP_URL"); v != "" {
    		return v
    	}
    	return "amqp://guest:guest@localhost:5672/"
    }

    func main() {
    	conn, err := amqp.Dial(amqpURL())
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	defer func() { _ = conn.Close() }()

    	ch, err := conn.Channel()
    	if err != nil {
    		log.Fatalf("channel: %v", err)
    	}
    	defer func() { _ = ch.Close() }()

    	// Declare "hello" → KubeMQ Queue channel amqp.default.hello.
    	q, err := ch.QueueDeclare("hello", false, false, false, false, nil)
    	if err != nil {
    		log.Fatalf("declare queue: %v", err)
    	}

    	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
    	defer cancel()

    	// Publish on the default exchange — routing key = queue name.
    	if err := ch.PublishWithContext(ctx, "", q.Name, false, false, amqp.Publishing{
    		ContentType: "text/plain",
    		Body:        []byte("Hello World!"),
    	}); err != nil {
    		log.Fatalf("publish: %v", err)
    	}
    	log.Printf(" [x] Sent %q", "Hello World!")

    	// Consume with auto-ack and print the exact body.
    	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
    	if err != nil {
    		log.Fatalf("consume: %v", err)
    	}
    	select {
    	case d := <-msgs:
    		log.Printf(" [x] Received %q", string(d.Body))
    	case <-ctx.Done():
    		log.Fatalf("timed out: %v", ctx.Err())
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os

    import pika

    URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
    QUEUE = "hello"  # → KubeMQ Queue channel amqp.default.hello


    def main() -> None:
        connection = pika.BlockingConnection(pika.URLParameters(URL))
        channel = connection.channel()

        # Default (nameless) exchange: routing key == queue name.
        channel.queue_declare(queue=QUEUE, durable=False, exclusive=False, auto_delete=False)

        channel.basic_publish(
            exchange="",
            routing_key=QUEUE,
            body=b"Hello World!",
            properties=pika.BasicProperties(content_type="text/plain"),
        )
        print(" [x] Sent 'Hello World!'")

        method, _props, body = channel.basic_get(queue=QUEUE, auto_ack=True)
        if method is None:
            raise SystemExit("no message received from queue")
        print(f" [x] Received {body.decode()!r}")

        channel.close()
        connection.close()


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DeliverCallback;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;

    public final class Main {

        private static final String QUEUE = "hello"; // → amqp.default.hello

        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(System.getenv().getOrDefault(
                    "KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"));

            try (Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel()) {

                // queue.declare: non-durable, non-exclusive, no auto-delete.
                channel.queueDeclare(QUEUE, false, false, false, null);

                // Publish via the default exchange ("") with routing key = queue name.
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .contentType("text/plain")
                        .build();
                channel.basicPublish("", QUEUE, props, "Hello World!".getBytes("UTF-8"));
                System.out.println("[x] Sent 'Hello World!'");

                BlockingQueue<String> received = new ArrayBlockingQueue<>(1);
                DeliverCallback onDeliver = (tag, delivery) ->
                        received.offer(new String(delivery.getBody(), "UTF-8"));
                channel.basicConsume(QUEUE, true, onDeliver, tag -> { });

                String got = received.poll(15, TimeUnit.SECONDS);
                if (got == null) {
                    throw new IllegalStateException("timed out waiting for the message");
                }
                System.out.println("[x] Received '" + got + "'");
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```javascript
    import amqp from "amqplib";

    // amqplib reads the URL path as the vhost; the bare trailing "/" in the dev URL
    // resolves to the default vhost (KubeMQ segment "default").
    const URL = process.env.KUBEMQ_AMQP_URL ?? "amqp://guest:guest@localhost:5672/";
    const QUEUE = "hello"; // → KubeMQ Queue channel amqp.default.hello

    async function main() {
      const conn = await amqp.connect(URL);
      const ch = await conn.createChannel();

      await ch.assertQueue(QUEUE, { durable: false, autoDelete: false, exclusive: false });

      const received = new Promise((resolve) => {
        ch.consume(
          QUEUE,
          (msg) => {
            if (msg === null) return;
            console.log(`[x] Received: ${msg.content.toString()}`);
            resolve();
          },
          { noAck: true },
        );
      });

      // Default exchange ("") routes by queue name.
      ch.publish("", QUEUE, Buffer.from("Hello World!"), { contentType: "text/plain" });
      console.log("[x] Sent:     Hello World!");

      await received;
      await ch.close();
      await conn.close();
    }

    main().catch((err) => {
      console.error("publish failed:", err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using System.Text;
    using RabbitMQ.Client;
    using RabbitMQ.Client.Events;

    const string queueName = "hello"; // → KubeMQ Queue channel amqp.default.hello

    var factory = new ConnectionFactory
    {
        Uri = new Uri(Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL")
                      ?? "amqp://guest:guest@localhost:5672/"),
    };

    using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
    var ct = cts.Token;

    await using var connection = await factory.CreateConnectionAsync(ct);
    await using var channel = await connection.CreateChannelAsync(cancellationToken: ct);

    // Non-durable, not exclusive, not auto-delete — a plain shared queue.
    await channel.QueueDeclareAsync(queueName, durable: false, exclusive: false,
        autoDelete: false, arguments: null, cancellationToken: ct);

    var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
    var consumer = new AsyncEventingBasicConsumer(channel);
    consumer.ReceivedAsync += (_, ea) =>
    {
        received.TrySetResult(Encoding.UTF8.GetString(ea.Body.Span));
        return Task.CompletedTask;
    };
    await channel.BasicConsumeAsync(queueName, autoAck: true, consumer: consumer, cancellationToken: ct);

    // Publish to the default exchange with routing key = queue name.
    var props = new BasicProperties { ContentType = "text/plain" };
    await channel.BasicPublishAsync(exchange: "", routingKey: queueName, mandatory: false,
        basicProperties: props, body: Encoding.UTF8.GetBytes("Hello World!"), cancellationToken: ct);
    Console.WriteLine("[x] Sent 'Hello World!'");

    var message = await received.Task.WaitAsync(ct);
    Console.WriteLine($"[x] Received '{message}'");
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # frozen_string_literal: true
    require "bunny"

    URL = ENV.fetch("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
    QUEUE = "hello" # → KubeMQ Queue channel amqp.default.hello

    conn = Bunny.new(URL)
    conn.start
    ch = conn.create_channel

    # Publisher confirms: wait for the broker to accept and route the publish before
    # reading it back, so a fire-and-forget publish is never lost in flight.
    ch.confirm_select

    queue = ch.queue(QUEUE, durable: false, auto_delete: false, exclusive: false)

    # Default ("") exchange routes by queue name.
    ch.default_exchange.publish("Hello World!", routing_key: queue.name, content_type: "text/plain")
    ch.wait_for_confirms
    puts " [x] Sent 'Hello World!'"

    body = nil
    deadline = Time.now + 5
    while body.nil? && Time.now < deadline
      _info, _props, body = queue.pop(manual_ack: false)
      sleep 0.2 if body.nil?
    end
    puts " [x] Received '#{body}'"

    conn.close
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use futures_lite::StreamExt;
    use lapin::{
        options::{BasicConsumeOptions, BasicPublishOptions, QueueDeclareOptions},
        types::FieldTable,
        BasicProperties, Connection, ConnectionProperties,
    };

    const QUEUE: &str = "hello"; // → KubeMQ Queue channel amqp.default.hello

    // lapin reads the URL path as the vhost and treats a bare trailing "/" as an
    // empty vhost, which the connector rejects. The default "/" vhost must be
    // percent-encoded as "%2f".
    fn amqp_url() -> String {
        let url = std::env::var("KUBEMQ_AMQP_URL")
            .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/".into());
        let host = url.rsplit_once('@').map_or(url.as_str(), |(_, h)| h);
        if host.ends_with('/') && !host.ends_with("/%2f") {
            format!("{url}%2f")
        } else {
            url
        }
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let conn = Connection::connect(&amqp_url(), ConnectionProperties::default()).await?;
        let channel = conn.create_channel().await?;

        channel
            .queue_declare(QUEUE, QueueDeclareOptions::default(), FieldTable::default())
            .await?;

        // Publish to the default exchange, routing key = queue name.
        channel
            .basic_publish(
                "",
                QUEUE,
                BasicPublishOptions::default(),
                b"Hello World!",
                BasicProperties::default().with_content_type("text/plain".into()),
            )
            .await?
            .await?;
        println!("[x] Sent 'Hello World!'");

        let mut consumer = channel
            .basic_consume(
                QUEUE,
                "hello-consumer",
                BasicConsumeOptions { no_ack: true, ..Default::default() },
                FieldTable::default(),
            )
            .await?;

        if let Some(delivery) = consumer.next().await {
            let delivery = delivery?;
            println!("[x] Received '{}'", String::from_utf8_lossy(&delivery.data));
        }

        conn.close(0, "done").await?;
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Supported languages [#supported-languages]

The connector speaks standard AMQP 0-9-1, so any conformant RabbitMQ client works. The
examples pin one native client per language — there is no KubeMQ SDK, no proto bindings,
and no published package.

| Language                | Client library                                                                         | Notes                                               |
| ----------------------- | -------------------------------------------------------------------------------------- | --------------------------------------------------- |
| Go                      | [`github.com/rabbitmq/amqp091-go`](https://github.com/rabbitmq/amqp091-go)             | The RabbitMQ team's Go client.                      |
| Python                  | [`pika`](https://pika.readthedocs.io/)                                                 | `BlockingConnection` with `URLParameters`.          |
| Java                    | [`com.rabbitmq:amqp-client`](https://www.rabbitmq.com/client-libraries/java-api-guide) | The official RabbitMQ Java client.                  |
| JavaScript / TypeScript | [`amqplib`](https://github.com/amqp-node/amqplib)                                      | Encode the default `/` vhost as `%2f` in the URL.   |
| C# / .NET               | [`RabbitMQ.Client`](https://www.rabbitmq.com/client-libraries/dotnet-api-guide)        | Task-based async API (v7+).                         |
| Ruby                    | [`bunny`](https://github.com/ruby-amqp/bunny)                                          | Use publisher confirms before consuming.            |
| Rust                    | [`lapin`](https://github.com/amqp-rs/lapin)                                            | async/await on Tokio; percent-encode the `/` vhost. |

## Next steps [#next-steps]

<Cards>
  <Card title="Getting started" href="/connectors/rabbitmq/tutorials/getting-started" description="Connect, declare, publish, and consume a message end-to-end through the RabbitMQ connector in minutes." />

  <Card title="Configuration" href="/connectors/rabbitmq/concepts/configuration" description="The 12 connector settings, the CONNECTORS_AMQP_ENABLE disable var, and the reserved default vhost." />

  <Card title="Work queues" href="/connectors/rabbitmq/how-to/work-queues" description="Competing consumers and fair dispatch over a single KubeMQ Queue channel." />

  <Card title="Channel mapping" href="/connectors/rabbitmq/reference/channel-mapping" description="The amqp.{vhost}.{queue} grammar, name constraints, and the property/header mapping." />
</Cards>