# Pub/Sub (Fanout) (/connectors/rabbitmq/how-to/pub-sub)



**Publish/subscribe** broadcasts every message to **all** interested subscribers. In AMQP this is a **fanout** exchange: every queue bound to the exchange receives a copy, and the routing key is ignored. Each subscriber typically declares its **own** server-named, exclusive queue, so subscribers are independent and their queues vanish when they disconnect. Each of those queues is an ordinary KubeMQ Queue channel — the exchange itself is **virtual connector-side routing** resolved at publish time, not a data store.

## Overview [#overview]

The publisher declares a **fanout** exchange and publishes to it; the routing key is ignored. Each subscriber declares a **server-named exclusive** queue (`queue.declare("")` → an `amq.gen-*` name) and binds it to the exchange with an empty routing key. At publish time the connector resolves the fanout into the set of bound queues and writes a copy to each one's KubeMQ channel — so every subscriber gets its own copy.

| Operation        | AMQP action                                       | KubeMQ mapping                                      |
| ---------------- | ------------------------------------------------- | --------------------------------------------------- |
| Declare exchange | `exchange.declare("logs", "fanout")`              | Virtual fanout routing entry (no storage)           |
| Subscribe        | `queue.declare("")` + `queue.bind(q, "logs", "")` | Server-named queue channel `amqp.default.amq.gen-*` |
| Publish          | `basic.publish(exchange="logs", routing-key="")`  | Copy fanned out to every bound queue channel        |
| Receive          | `basic.consume(q)` (auto-ack)                     | One copy per subscriber                             |

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

A published message fans out to every queue bound to the exchange. Because each subscriber binds its own exclusive queue, every subscriber receives every message independently — there is no competition between them.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher<br/>(publish → logs)&#x22;]
CONN[&#x22;RabbitMQ connector<br/>:5672&#x22;]
X([&#x22;fanout exchange<br/>logs (virtual)&#x22;])
BROKER[&#x22;Message Broker&#x22;]
QA{{&#x22;amq.gen-A&#x22;}}
QB{{&#x22;amq.gen-B&#x22;}}
S1[&#x22;Subscriber A&#x22;]
S2[&#x22;Subscriber B&#x22;]

PUB -- &#x22;basic.publish (key ignored)&#x22; --> CONN
CONN -- &#x22;resolve fanout&#x22; --> X
X -- &#x22;copy&#x22; --> QA
X -- &#x22;copy&#x22; --> QB
QA --> BROKER
QB --> BROKER
BROKER -. &#x22;deliver&#x22; .-> S1
BROKER -. &#x22;deliver&#x22; .-> S2

class PUB,S1,S2 client
class CONN connector
class X queue
class BROKER broker
class QA,QB queue
`"
/>

*The virtual fanout exchange resolves to every bound queue; the connector writes a copy to each queue channel, so each subscriber receives its own copy.*

## Publish and subscribe [#publish-and-subscribe]

Each example declares a non-durable fanout exchange `logs`, attaches **two** subscribers (each with its own server-named exclusive queue bound with the empty key), then broadcasts a batch of messages on a confirm channel. Each subscriber independently receives every message. Every client reads the broker endpoint from `KUBEMQ_AMQP_URL` (default `amqp://guest:guest@localhost:5672/`).

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

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

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

    const exchange = "logs"
    const messages = 5

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

    func subscribe(conn *amqp.Connection) (string, <-chan amqp.Delivery) {
    	ch, err := conn.Channel()
    	if err != nil {
    		log.Fatalf("subscriber channel: %v", err)
    	}
    	if err := ch.ExchangeDeclare(exchange, "fanout", false, false, false, false, nil); err != nil {
    		log.Fatalf("declare exchange: %v", err)
    	}
    	q, err := ch.QueueDeclare("", false, false, true, false, nil) // server-named, exclusive
    	if err != nil {
    		log.Fatalf("declare queue: %v", err)
    	}
    	if err := ch.QueueBind(q.Name, "", exchange, false, nil); err != nil { // key ignored
    		log.Fatalf("bind: %v", err)
    	}
    	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil) // auto-ack
    	if err != nil {
    		log.Fatalf("consume: %v", err)
    	}
    	return q.Name, msgs
    }

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

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

    	nameA, subA := subscribe(conn)
    	nameB, subB := subscribe(conn)
    	log.Printf(" [s] bound exclusive queues %q and %q to fanout %q", nameA, nameB, exchange)

    	// Publish on a confirm channel so every broadcast is accepted before we stop.
    	pubCh, _ := conn.Channel()
    	if err := pubCh.Confirm(false); err != nil {
    		log.Fatalf("confirm select: %v", err)
    	}
    	confirms := pubCh.NotifyPublish(make(chan amqp.Confirmation, messages))
    	for i := 0; i < messages; i++ {
    		if err := pubCh.PublishWithContext(ctx, exchange, "", false, false, amqp.Publishing{
    			ContentType: "text/plain",
    			Body:        []byte(fmt.Sprintf("log-%d", i)),
    		}); err != nil {
    			log.Fatalf("publish %d: %v", i, err)
    		}
    	}
    	for i := 0; i < messages; i++ {
    		<-confirms
    	}
    	log.Printf(" [x] Broadcast %d messages to fanout %q", messages, exchange)

    	// Each subscriber independently receives all 5 copies.
    	for label, msgs := range map[string]<-chan amqp.Delivery{"A": subA, "B": subB} {
    		got := 0
    		for got < messages {
    			select {
    			case <-msgs:
    				got++
    			case <-ctx.Done():
    				log.Fatalf("subscriber %s timed out (%d/%d)", label, got, messages)
    			}
    		}
    		log.Printf(" [s%s] received all %d broadcasts", label, got)
    	}
    }
    ```
  </Tab>

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

    import pika

    URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
    EXCHANGE = "logs"
    SUBSCRIBERS = 2
    MESSAGES = 5


    def main() -> None:
        # Subscribers — each on its own connection with a server-named exclusive
        # queue bound to the fanout exchange.
        subs = []
        for _ in range(SUBSCRIBERS):
            conn = pika.BlockingConnection(pika.URLParameters(URL))
            ch = conn.channel()
            ch.exchange_declare(exchange=EXCHANGE, exchange_type="fanout", durable=False)
            queue = ch.queue_declare(queue="", exclusive=True).method.queue  # amq.gen-*
            ch.queue_bind(exchange=EXCHANGE, queue=queue, routing_key="")  # key ignored
            print(f" [sub] bound exclusive queue {queue!r}")
            subs.append((conn, ch, queue))

        # Publisher — broadcast 5 messages on a confirm channel.
        pub_conn = pika.BlockingConnection(pika.URLParameters(URL))
        pub_ch = pub_conn.channel()
        pub_ch.exchange_declare(exchange=EXCHANGE, exchange_type="fanout", durable=False)
        pub_ch.confirm_delivery()
        for i in range(MESSAGES):
            pub_ch.basic_publish(exchange=EXCHANGE, routing_key="", body=f"log-{i}".encode())
        print(f" [x] Broadcast {MESSAGES} messages to {EXCHANGE!r}")

        # Each subscriber independently receives every copy.
        for idx, (conn, ch, queue) in enumerate(subs):
            got = 0
            for method, _props, _body in ch.consume(queue, inactivity_timeout=30, auto_ack=True):
                if method is None:
                    raise SystemExit(f"subscriber {idx}: timed out ({got}/{MESSAGES})")
                got += 1
                if got >= MESSAGES:
                    break
            ch.cancel()
            print(f" [sub {idx}] received all {got} broadcasts")
            ch.close()
            conn.close()

        pub_ch.close()
        pub_conn.close()


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

  <Tab value="Java">
    ```java
    import java.nio.charset.StandardCharsets;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;

    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;

    public final class Main {
        private static final String EXCHANGE = "logs";
        private static final int SUBSCRIBERS = 2;
        private static final int MESSAGES = 5;

        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/"));
            if (factory.getVirtualHost() == null || factory.getVirtualHost().isEmpty()) {
                factory.setVirtualHost("/"); // trailing "/" parses to an empty vhost
            }

            Connection connection = factory.newConnection();

            // Subscribers — each with a server-named exclusive queue bound to fanout.
            List<CountDownLatch> latches = new ArrayList<>();
            for (int i = 0; i < SUBSCRIBERS; i++) {
                Channel ch = connection.createChannel();
                ch.exchangeDeclare(EXCHANGE, "fanout", false);
                String queue = ch.queueDeclare("", false, true, true, null).getQueue(); // amq.gen-*
                ch.queueBind(queue, EXCHANGE, ""); // key ignored for fanout
                CountDownLatch latch = new CountDownLatch(MESSAGES);
                ch.basicConsume(queue, true, (tag, delivery) -> latch.countDown(), tag -> { });
                System.out.println("[sub] bound exclusive queue " + queue);
                latches.add(latch);
            }

            // Publisher — broadcast 5 messages on a confirm channel.
            Channel pub = connection.createChannel();
            pub.exchangeDeclare(EXCHANGE, "fanout", false);
            pub.confirmSelect();
            for (int i = 0; i < MESSAGES; i++) {
                pub.basicPublish(EXCHANGE, "", null, ("log-" + i).getBytes(StandardCharsets.UTF_8));
            }
            pub.waitForConfirmsOrDie(30_000);
            System.out.println("[x] Broadcast " + MESSAGES + " messages to fanout '" + EXCHANGE + "'");

            // Each subscriber independently receives all 5 copies.
            for (int i = 0; i < latches.size(); i++) {
                if (!latches.get(i).await(30, TimeUnit.SECONDS)) {
                    throw new IllegalStateException("subscriber " + i + " did not receive all broadcasts");
                }
                System.out.println("[sub " + i + "] received all " + MESSAGES + " broadcasts");
            }
            connection.close();
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import amqp, { type Channel, type ChannelModel } from "amqplib";

    const EXCHANGE = "logs";
    const MESSAGES = ["log-0", "log-1", "log-2", "log-3", "log-4"];

    function url(): string {
      return process.env["KUBEMQ_AMQP_URL"] ?? "amqp://guest:guest@localhost:5672/";
    }

    async function subscribe(connection: ChannelModel): Promise<{ name: string; done: Promise<void> }> {
      const ch: Channel = await connection.createChannel();
      await ch.assertExchange(EXCHANGE, "fanout", { durable: false });
      const q = await ch.assertQueue("", { exclusive: true }); // amq.gen-*
      await ch.bindQueue(q.queue, EXCHANGE, ""); // routing key ignored for fanout
      let received = 0;
      const done = new Promise<void>((resolve) => {
        ch.consume(q.queue, (msg) => {
          if (msg === null) return;
          if (++received === MESSAGES.length) resolve();
        }, { noAck: true });
      });
      return { name: q.queue, done };
    }

    async function main(): Promise<void> {
      const connection = await amqp.connect(url());

      const subA = await subscribe(connection);
      const subB = await subscribe(connection);
      console.log(`[sub] bound exclusive queues ${subA.name} and ${subB.name}`);

      // Publisher — broadcast on a confirm channel so no message is lost to an early close.
      const pubCh = await connection.createConfirmChannel();
      await pubCh.assertExchange(EXCHANGE, "fanout", { durable: false });
      for (const body of MESSAGES) pubCh.publish(EXCHANGE, "", Buffer.from(body));
      await pubCh.waitForConfirms();
      console.log(`[x] Broadcast ${MESSAGES.length} messages to fanout "${EXCHANGE}"`);

      await Promise.all([subA.done, subB.done]);
      console.log("both subscribers received all 5 broadcasts independently");
      await connection.close();
    }

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

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

    const string exchange = "logs";
    const int subscribers = 2;
    const int messages = 5;

    static string Url() =>
        Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL") is { Length: > 0 } v
            ? v
            : "amqp://guest:guest@localhost:5672/";

    var factory = new ConnectionFactory { Uri = new Uri(Url()) };
    await using var connection = await factory.CreateConnectionAsync("pub-sub-fanout");

    // Subscribers — each with a server-named exclusive queue bound to the fanout.
    var latches = new List<TaskCompletionSource>();
    for (var i = 0; i < subscribers; i++)
    {
        var ch = await connection.CreateChannelAsync();
        await ch.ExchangeDeclareAsync(exchange, ExchangeType.Fanout, durable: false);
        var queue = (await ch.QueueDeclareAsync("", durable: false, exclusive: true, autoDelete: true)).QueueName;
        await ch.QueueBindAsync(queue, exchange, ""); // routing key ignored for fanout
        var done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var got = 0;
        var consumer = new AsyncEventingBasicConsumer(ch);
        consumer.ReceivedAsync += (_, _) =>
        {
            if (Interlocked.Increment(ref got) == messages) done.TrySetResult();
            return Task.CompletedTask;
        };
        await ch.BasicConsumeAsync(queue, autoAck: true, consumer: consumer);
        Console.WriteLine($"[sub] bound exclusive queue {queue}");
        latches.Add(done);
    }

    // Publisher — broadcast on a confirm channel.
    var confirmOpts = new CreateChannelOptions(
        publisherConfirmationsEnabled: true,
        publisherConfirmationTrackingEnabled: true);
    await using var pub = await connection.CreateChannelAsync(confirmOpts);
    await pub.ExchangeDeclareAsync(exchange, ExchangeType.Fanout, durable: false);
    for (var i = 0; i < messages; i++)
        await pub.BasicPublishAsync(exchange, "", body: Encoding.UTF8.GetBytes($"log-{i}"));
    Console.WriteLine($"[x] Broadcast {messages} messages to fanout '{exchange}'");

    using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
    await Task.WhenAll(latches.Select(l => l.Task.WaitAsync(cts.Token)));
    Console.WriteLine("both subscribers received all broadcasts independently");
    ```
  </Tab>

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

    EXCHANGE = "logs"
    MESSAGE_COUNT = 5

    opts = AMQ::URI.parse(ENV.fetch("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"))
    opts[:vhost] = "/" if opts[:vhost].nil? || opts[:vhost].to_s.empty?
    conn = Bunny.new(opts)
    conn.start

    pub_ch = conn.create_channel
    pub_ch.confirm_select
    exchange = pub_ch.fanout(EXCHANGE, durable: false)

    # Two subscribers, each with its own server-named exclusive queue.
    subscribers = Array.new(2) do |idx|
      sub_ch = conn.create_channel
      queue = sub_ch.queue("", exclusive: true) # amq.gen-*
      queue.bind(exchange) # routing key ignored for fanout
      received = Queue.new
      queue.subscribe(manual_ack: false, block: false) { |_di, _props, _body| received.push(:msg) }
      puts " [*] Subscriber #{idx + 1} bound exclusive queue #{queue.name}"
      { id: idx + 1, received: received }
    end

    # Broadcast 5 messages, then wait for confirms.
    MESSAGE_COUNT.times { |i| exchange.publish("log-#{i}", routing_key: "ignored") }
    pub_ch.wait_for_confirms
    puts " [x] Broadcast #{MESSAGE_COUNT} messages to fanout '#{EXCHANGE}'"

    # Each subscriber independently receives all 5.
    subscribers.each do |s|
      MESSAGE_COUNT.times { s[:received].pop }
      puts " [x] Subscriber #{s[:id]} received all #{MESSAGE_COUNT} broadcasts"
    end

    exchange.delete
    conn.close
    ```
  </Tab>

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

    const EXCHANGE: &str = "logs";
    const MESSAGES: usize = 5;

    fn amqp_url() -> String {
        let url = std::env::var("KUBEMQ_AMQP_URL")
            .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/".into());
        match url.rsplit_once('@').map_or(url.as_str(), |(_, host)| host) {
            host if host.ends_with('/') && !host.to_lowercase().ends_with("/%2f") => format!("{url}%2f"),
            _ => url,
        }
    }

    async fn subscribe(conn: &Connection) -> Result<lapin::Consumer, Box<dyn std::error::Error>> {
        let ch = conn.create_channel().await?;
        ch.exchange_declare(EXCHANGE, ExchangeKind::Fanout, ExchangeDeclareOptions::default(), FieldTable::default())
            .await?;
        let queue = ch
            .queue_declare("", QueueDeclareOptions { exclusive: true, ..Default::default() }, FieldTable::default())
            .await?;
        ch.queue_bind(queue.name().as_str(), EXCHANGE, "", QueueBindOptions::default(), FieldTable::default())
            .await?;
        let consumer = ch
            .basic_consume(queue.name().as_str(), "", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
            .await?;
        println!("[sub] bound exclusive queue {}", queue.name());
        Ok(consumer)
    }

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

        let mut sub_a = subscribe(&conn).await?;
        let mut sub_b = subscribe(&conn).await?;

        // Publisher — broadcast on a confirm channel.
        let pub_ch = conn.create_channel().await?;
        pub_ch.exchange_declare(EXCHANGE, ExchangeKind::Fanout, ExchangeDeclareOptions::default(), FieldTable::default())
            .await?;
        pub_ch.confirm_select(ConfirmSelectOptions::default()).await?;
        for i in 0..MESSAGES {
            pub_ch
                .basic_publish(EXCHANGE, "", BasicPublishOptions::default(), format!("log-{i}").as_bytes(), BasicProperties::default())
                .await?
                .await?;
        }
        println!("[x] Broadcast {MESSAGES} messages to fanout '{EXCHANGE}'");

        for (label, consumer) in [("A", &mut sub_a), ("B", &mut sub_b)] {
            for _ in 0..MESSAGES {
                consumer.next().await.ok_or("subscriber stream closed early")??;
            }
            println!("[s{label}] received all {MESSAGES} broadcasts");
        }
        conn.close(0, "done").await?;
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Server-named exclusive queues [#server-named-exclusive-queues]

`queue.declare("")` makes the broker mint a unique name (`amq.gen-{id}`). Declaring it **exclusive** scopes it to the subscriber's connection and auto-deletes it on disconnect — exactly what you want for a transient subscriber. Each such queue is a normal KubeMQ Queue channel that lives only as long as the subscriber.

<Callout type="warn">
  **Confirm before closing a broadcast producer.** Without publisher confirms, `basic.publish` only buffers the message on the connector's per-channel executor and returns; closing the channel or connection before that buffer drains **silently abandons** the un-ingested publishes — no error, no nack. A tight broadcast-then-close loop can lose most of a batch. Enable `confirm.select` (a confirm channel) and **wait for all acks before closing**, as every example above does, or keep the connection open until subscribers have drained.
</Callout>

<Callout type="info">
  **Exclusive queues are node-local in a cluster.** An exclusive (server-named) queue lives only on the node owning the subscriber's connection; the publisher must reach the same node for the broadcast to land. Single-node deployments are unaffected.
</Callout>

## Related [#related]

<Cards>
  <Card title="Routing (direct)" href="/connectors/rabbitmq/how-to/routing" description="Selective delivery — only queues whose binding key matches exactly receive the message." />

  <Card title="Topics" href="/connectors/rabbitmq/how-to/topics" description="Pattern routing with * and # wildcards over dot-separated routing keys." />

  <Card title="Channel mapping" href="/connectors/rabbitmq/reference/channel-mapping" description="How exchanges, bindings, and queues resolve to KubeMQ Queue channels." />
</Cards>