# Getting Started (/connectors/rabbitmq/tutorials/getting-started)



Get a message flowing through the KubeMQ RabbitMQ (AMQP 0-9-1) connector in minutes. You
point a standard RabbitMQ client at the broker, declare a queue, publish a message through
the default exchange, and consume it back — all over the native AMQP 0-9-1 wire, with no
KubeMQ SDK. This walkthrough takes you from a running server to a verified round-trip.

## Prerequisites [#prerequisites]

* A running **kubemq-server** with the RabbitMQ connector **enabled** and reachable on **port 5672** (plain
  TCP). The connector is **opt-in (disabled by default)** — see the enable step below.
* One of the AMQP 0-9-1 clients below for your language (the examples pin a native RabbitMQ
  client per language — there is no KubeMQ SDK).

## Enable the connector [#enable-the-connector]

The RabbitMQ (AMQP 0-9-1) connector is **disabled by default** — a stock kubemq-server does
**not** bind the AMQP listener until you turn it on. Enable it with its enable variable:

<RunKubeMQ ports="[5672, 5671, 50000]" env="{ CONNECTORS_AMQP_ENABLE: 'true' }" />

<Callout type="warn">
  **The enable variable is `CONNECTORS_AMQP_ENABLE`** — note that this is the &#x2A;*AMQP 0-9-1
  (RabbitMQ)** connector, distinct from the AMQP 1.0 connector's `CONNECTORS_AMQP10_ENABLE`.
  For Kubernetes, set `spec.amqp.enabled: true` in the `KubemqCluster` CR.
</Callout>

Bring up a throwaway local broker with the RabbitMQ connector enabled:

Every example reads a single environment variable for the broker endpoint. On a development
broker, auth is disabled and `guest:guest` is accepted with no JWT:

```bash
# default: amqp://guest:guest@localhost:5672/
export KUBEMQ_AMQP_URL="amqp://guest:guest@localhost:5672/"
```

The AMQP vhost `/` maps to the connector's configured `DefaultVhost` segment (literal
`"default"`), so queue `hello` on vhost `/` lands on the KubeMQ Queue channel
`amqp.default.hello`. Connect to vhost `/`, not to a literal vhost named `default` — the
`default` segment is reserved and a direct connection to it is rejected. See
[Channel mapping](/connectors/rabbitmq/reference/channel-mapping).

To **disable** the RabbitMQ connector after enabling it, set its enable variable to `false`:

<RunKubeMQ variant="disable" env="{ CONNECTORS_AMQP_ENABLE: 'false' }" />

AMQP 0-9-1 and AMQP 1.0 share ports 5672/5671 but have separate enable flags — disabling
RabbitMQ leaves AMQP 1.0 reachable on the same ports, and vice-versa. You can also disable
just one listener by setting `CONNECTORS_AMQP_PORT=0` (plain) or `CONNECTORS_AMQP_TLS_PORT=0`
(TLS). See [Configuration](/connectors/rabbitmq/concepts/configuration) for the full settings list.

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

A publisher declares a queue and publishes to an exchange with a routing key. The connector
resolves the exchange routing to the target queue at publish time and writes the message to
that queue's KubeMQ Queue channel `amqp.{vhost}.{queue}`. A consumer subscribed to the same
queue receives it.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher<br/>basic.publish('', 'hello')&#x22;]
CONN[&#x22;RabbitMQ connector<br/>:5672&#x22;]
CH{{&#x22;Queue channel<br/>amqp.default.hello&#x22;}}
SUB[&#x22;Consumer<br/>basic.consume('hello')&#x22;]

PUB -- &#x22;default exchange&#x22; --> CONN
CONN -- &#x22;(Queue, amqp.default.hello)&#x22; --> CH
CH -. &#x22;deliver&#x22; .-> CONN
CONN -. &#x22;basic.deliver&#x22; .-> SUB

class PUB,SUB client
class CONN connector
class CH queues
`"
/>

*Publishing to queue `hello` on the default exchange maps to the KubeMQ Queue channel `amqp.default.hello`; a consumer on the same queue receives the message.*

## Steps [#steps]

<Steps>
  <Step>
    ### Connect and declare a queue [#connect-and-declare-a-queue]

    Open a connection to the endpoint in `KUBEMQ_AMQP_URL`, open a channel, and declare the
    target queue. The language tabs run the **complete** round-trip from a single program:
    connect, declare the `hello` queue, publish one message through the default exchange, and
    consume it back.

    <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.Println(" [x] Sent '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()) {

                    channel.queueDeclare(QUEUE, false, false, false, null);

                    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";

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

          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("getting-started 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);

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

        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
        ch.confirm_select # publisher confirms, so the publish is routed before we consume

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

        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

        // The default "/" vhost must be percent-encoded as "%2f" for lapin's URI parser.
        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?;

            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>
  </Step>

  <Step>
    ### Publish a message [#publish-a-message]

    The program above publishes one `text/plain` message to the **default exchange** with the
    routing key set to the queue name `hello`. The default (nameless) exchange routes a message
    to the queue whose name equals the routing key, so the message lands in queue `hello` — the
    KubeMQ Queue channel `amqp.default.hello`. Because a KubeMQ Queue is durable and
    at-least-once, you can publish before a consumer is attached and the message waits in the
    queue.
  </Step>

  <Step>
    ### Consume and verify [#consume-and-verify]

    A consumer subscribed to `hello` receives the message. When it arrives the program prints
    it and exits:

    ```text
     [x] Sent 'Hello World!'
     [x] Received 'Hello World!'
    ```

    Unlike fire-and-forget pub/sub, the Queue holds the message until a consumer acknowledges
    it, so order of operations is forgiving. For competing consumers and fair dispatch across a
    worker pool, see [Work queues](/connectors/rabbitmq/how-to/work-queues).
  </Step>
</Steps>

<Callout type="info">
  Exchanges and bindings are **virtual** — resolved at publish time, not stored. A `fanout`,
  `direct`, `topic`, or `headers` exchange routes to a set of queues, and each resolved queue
  is a normal KubeMQ Queue channel. See
  [Exchanges and routing](/connectors/rabbitmq/concepts/exchanges-and-routing).
</Callout>

## Next steps [#next-steps]

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

  <Card title="Architecture" href="/connectors/rabbitmq/concepts/architecture" description="The everything-is-a-Queue model, virtual exchanges, and cross-protocol interop." />

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

  <Card title="Pub/sub" href="/connectors/rabbitmq/how-to/pub-sub" description="Fanout exchanges fan a publish out to every bound queue at publish time." />
</Cards>