# Topics (/connectors/rabbitmq/how-to/topics)



A **topic** exchange routes by **pattern**. Routing keys are dot-separated words (`stock.usd.nyse`), and bindings use wildcards to match families of keys. It generalizes the [direct exchange](/connectors/rabbitmq/how-to/routing) (exact match) into flexible, hierarchical routing. Like every exchange here, the topic exchange is **virtual connector-side routing**: at publish time the connector matches the key against the bindings and writes a copy to each matched queue's KubeMQ channel.

## Overview [#overview]

Bindings use two wildcards over the `.` word separator:

| Token | Matches                |
| ----- | ---------------------- |
| `*`   | exactly **one** word   |
| `#`   | **zero or more** words |

| Binding      | Matches                                      | Does NOT match                        |
| ------------ | -------------------------------------------- | ------------------------------------- |
| `*.orange.*` | `quick.orange.rabbit`, `lazy.orange.fox`     | `lazy.orange` (needs two words after) |
| `*.*.rabbit` | `quick.orange.rabbit`, `lazy.pink.rabbit`    | `rabbit`                              |
| `lazy.#`     | `lazy`, `lazy.brown.fox`, `lazy.pink.rabbit` | `quick.brown.fox`                     |
| `#`          | any key, including the empty key             | —                                     |

If one queue is bound with **multiple** patterns that both match a key, the message is delivered to that queue exactly **once** — matched queues are deduplicated.

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

Each publish is matched against every binding; the resulting set of queues is deduplicated, so a queue that matches a key through two bindings still receives a single copy. A key that matches no binding is dropped.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher<br/>(key=lazy.pink.rabbit)&#x22;]
CONN[&#x22;RabbitMQ connector<br/>:5672&#x22;]
X([&#x22;topic exchange<br/>topic_logs (virtual)&#x22;])
BROKER[&#x22;Message Broker&#x22;]
Q1{{&#x22;Q1<br/>(*.orange.*)&#x22;}}
Q2{{&#x22;Q2<br/>(*.*.rabbit + lazy.#)&#x22;}}

PUB -- &#x22;basic.publish&#x22; --> CONN
CONN -- &#x22;wildcard match&#x22; --> X
X -. &#x22;no orange word → no match&#x22; .-> Q1
X -- &#x22;matches BOTH Q2 bindings → ONE copy&#x22; --> Q2
Q2 --> BROKER

class PUB client
class CONN connector
class X queue
class BROKER broker
class Q1,Q2 queue
`"
/>

*The virtual topic exchange evaluates each binding pattern; `lazy.pink.rabbit` matches both of Q2's bindings yet is delivered to Q2 exactly once, while Q1's `*.orange.*` does not match.*

## Publish and match [#publish-and-match]

Each example declares a topic exchange `topic_logs`, binds Q1 on `*.orange.*` and Q2 on **both** `*.*.rabbit` and `lazy.#`, then publishes six keys. `lazy.pink.rabbit` matches both of Q2's bindings and arrives **once**; `quick.brown.fox` matches nobody and is dropped. 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"
    	"log"
    	"os"
    	"sort"
    	"time"

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

    const exchange = "topic_logs"

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

    func bindConsumer(conn *amqp.Connection, patterns ...string) <-chan amqp.Delivery {
    	ch, err := conn.Channel()
    	if err != nil {
    		log.Fatalf("channel: %v", err)
    	}
    	q, err := ch.QueueDeclare("", false, false, true, false, nil) // server-named, exclusive
    	if err != nil {
    		log.Fatalf("declare: %v", err)
    	}
    	for _, p := range patterns {
    		if err := ch.QueueBind(q.Name, p, exchange, false, nil); err != nil {
    			log.Fatalf("bind %s: %v", p, err)
    		}
    	}
    	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
    	if err != nil {
    		log.Fatalf("consume: %v", err)
    	}
    	return 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() }()

    	ch, _ := conn.Channel()
    	if err := ch.ExchangeDeclare(exchange, "topic", false, false, false, false, nil); err != nil {
    		log.Fatalf("declare exchange: %v", err)
    	}

    	q1 := bindConsumer(conn, "*.orange.*")          // orange animals
    	q2 := bindConsumer(conn, "*.*.rabbit", "lazy.#") // rabbits and everything lazy

    	keys := []string{
    		"quick.orange.rabbit",  // q1 + q2
    		"lazy.orange.elephant", // q1 + q2 (lazy.# spans two words)
    		"quick.orange.fox",     // q1 only
    		"lazy.brown.fox",       // q2 only (lazy.#)
    		"lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
    		"quick.brown.fox",      // nobody → silent drop
    	}
    	for _, key := range keys {
    		if err := ch.PublishWithContext(ctx, exchange, key, false, false, amqp.Publishing{
    			ContentType: "text/plain",
    			Body:        []byte(key),
    		}); err != nil {
    			log.Fatalf("publish %s: %v", key, err)
    		}
    	}
    	log.Printf(" [x] Published %d keys", len(keys))

    	collect := func(label string, msgs <-chan amqp.Delivery, want []string) {
    		seen := make(map[string]struct{}, len(want))
    		for len(seen) < len(want) {
    			select {
    			case d := <-msgs:
    				seen[string(d.Body)] = struct{}{}
    			case <-ctx.Done():
    				log.Fatalf("%s timed out (%d/%d)", label, len(seen), len(want))
    			}
    		}
    		out := make([]string, 0, len(seen))
    		for k := range seen {
    			out = append(out, k)
    		}
    		sort.Strings(out)
    		log.Printf(" [%s] matched: %v", label, out)
    	}
    	collect("q1", q1, []string{"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox"})
    	collect("q2", q2, []string{"quick.orange.rabbit", "lazy.orange.elephant", "lazy.brown.fox", "lazy.pink.rabbit"})
    	log.Printf(" [✓] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped")
    }
    ```
  </Tab>

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

    import pika

    URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
    EXCHANGE = "topic_logs"


    def main() -> None:
        conn = pika.BlockingConnection(pika.URLParameters(URL))
        ch = conn.channel()
        ch.exchange_declare(exchange=EXCHANGE, exchange_type="topic", durable=False)

        def bind(*patterns: str) -> str:
            queue = ch.queue_declare(queue="", exclusive=True).method.queue
            for p in patterns:
                ch.queue_bind(exchange=EXCHANGE, queue=queue, routing_key=p)
            return queue

        q1 = bind("*.orange.*")            # orange animals
        q2 = bind("*.*.rabbit", "lazy.#")  # rabbits and everything lazy

        keys = [
            "quick.orange.rabbit",   # q1 + q2
            "lazy.orange.elephant",  # q1 + q2
            "quick.orange.fox",      # q1 only
            "lazy.brown.fox",        # q2 only
            "lazy.pink.rabbit",      # q2 only — matches BOTH q2 bindings, ONE copy
            "quick.brown.fox",       # nobody → silent drop
        ]
        for key in keys:
            ch.basic_publish(exchange=EXCHANGE, routing_key=key, body=key.encode())
        print(f" [x] Published {len(keys)} keys")

        def collect(label: str, queue: str, want: set[str]) -> None:
            seen: set[str] = set()
            for method, _props, body in ch.consume(queue, inactivity_timeout=30, auto_ack=True):
                if method is None:
                    raise SystemExit(f"{label} timed out ({len(seen)}/{len(want)})")
                seen.add(body.decode())
                if len(seen) >= len(want):
                    break
            ch.cancel()
            print(f" [{label}] matched: {sorted(seen)}")

        collect("q1", q1, {"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox"})
        collect("q2", q2, {"quick.orange.rabbit", "lazy.orange.elephant", "lazy.brown.fox", "lazy.pink.rabbit"})
        print(" [x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped")
        ch.close()
        conn.close()


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

  <Tab value="Java">
    ```java
    import java.nio.charset.StandardCharsets;
    import java.util.Arrays;
    import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.ConcurrentHashMap;

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

    public final class Main {
        private static final String EXCHANGE = "topic_logs";

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

            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE, "topic", false);

            Set<String> q1 = ConcurrentHashMap.newKeySet();
            Set<String> q2 = ConcurrentHashMap.newKeySet();
            bind(connection, q1, "*.orange.*");
            bind(connection, q2, "*.*.rabbit", "lazy.#");

            String[] keys = {
                "quick.orange.rabbit",  // q1 + q2
                "lazy.orange.elephant", // q1 + q2
                "quick.orange.fox",     // q1 only
                "lazy.brown.fox",       // q2 only
                "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
                "quick.brown.fox",      // nobody → silent drop
            };
            for (String key : keys) {
                channel.basicPublish(EXCHANGE, key, null, key.getBytes(StandardCharsets.UTF_8));
            }
            System.out.println("[x] Published " + keys.length + " keys");

            await(q1, 3);
            await(q2, 4);
            System.out.println("[q1] matched: " + new HashSet<>(q1));
            System.out.println("[q2] matched: " + new HashSet<>(q2));
            System.out.println("[v] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");
            connection.close();
        }

        private static void bind(Connection connection, Set<String> sink, String... patterns) throws Exception {
            Channel ch = connection.createChannel();
            String queue = ch.queueDeclare("", false, true, true, null).getQueue();
            for (String p : patterns) {
                ch.queueBind(queue, EXCHANGE, p);
            }
            ch.basicConsume(queue, true,
                    (tag, d) -> sink.add(new String(d.getBody(), StandardCharsets.UTF_8)),
                    tag -> { });
        }

        private static void await(Set<String> sink, int expected) throws InterruptedException {
            long deadline = System.currentTimeMillis() + 30_000;
            while (sink.size() < expected && System.currentTimeMillis() < deadline) {
                Thread.sleep(50);
            }
            if (sink.size() < expected) {
                throw new IllegalStateException("expected " + expected + ", saw " + Arrays.toString(sink.toArray()));
            }
        }
    }
    ```
  </Tab>

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

    const EXCHANGE = "topic_logs";

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

    async function bind(connection: ChannelModel, want: number, ...patterns: string[]): Promise<Set<string>> {
      const ch = await connection.createChannel();
      const q = await ch.assertQueue("", { exclusive: true });
      for (const p of patterns) await ch.bindQueue(q.queue, EXCHANGE, p);
      const seen = new Set<string>();
      await ch.consume(q.queue, (msg) => {
        if (msg) seen.add(msg.content.toString());
      }, { noAck: true });
      // Returns the live set; the caller waits until it reaches `want`.
      return Object.assign(seen, { want });
    }

    async function main(): Promise<void> {
      const connection = await amqp.connect(url());
      const channel = await connection.createChannel();
      await channel.assertExchange(EXCHANGE, "topic", { durable: false });

      const q1 = await bind(connection, 3, "*.orange.*");
      const q2 = await bind(connection, 4, "*.*.rabbit", "lazy.#");

      const keys = [
        "quick.orange.rabbit",  // q1 + q2
        "lazy.orange.elephant", // q1 + q2
        "quick.orange.fox",     // q1 only
        "lazy.brown.fox",       // q2 only
        "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
        "quick.brown.fox",      // nobody → silent drop
      ];
      for (const key of keys) channel.publish(EXCHANGE, key, Buffer.from(key));
      console.log(`[x] Published ${keys.length} keys`);

      await waitFor(q1, 3);
      await waitFor(q2, 4);
      console.log(`[q1] matched: ${[...q1].sort()}`);
      console.log(`[q2] matched: ${[...q2].sort()}`);
      console.log("[x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");
      await connection.close();
    }

    function waitFor(set: Set<string>, n: number): Promise<void> {
      return new Promise((resolve, reject) => {
        const deadline = Date.now() + 30_000;
        const tick = setInterval(() => {
          if (set.size >= n) {
            clearInterval(tick);
            resolve();
          } else if (Date.now() > deadline) {
            clearInterval(tick);
            reject(new Error(`expected ${n}, saw ${set.size}`));
          }
        }, 50);
      });
    }

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

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

    const string exchange = "topic_logs";

    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("topics");
    await using var channel = await connection.CreateChannelAsync();
    await channel.ExchangeDeclareAsync(exchange, ExchangeType.Topic, durable: false);

    async Task<ConcurrentDictionary<string, byte>> Bind(params string[] patterns)
    {
        var ch = await connection.CreateChannelAsync();
        var queue = (await ch.QueueDeclareAsync("", durable: false, exclusive: true, autoDelete: true)).QueueName;
        foreach (var p in patterns) await ch.QueueBindAsync(queue, exchange, p);
        var seen = new ConcurrentDictionary<string, byte>();
        var consumer = new AsyncEventingBasicConsumer(ch);
        consumer.ReceivedAsync += (_, ea) =>
        {
            seen.TryAdd(Encoding.UTF8.GetString(ea.Body.Span), 0);
            return Task.CompletedTask;
        };
        await ch.BasicConsumeAsync(queue, autoAck: true, consumer: consumer);
        return seen;
    }

    var q1 = await Bind("*.orange.*");
    var q2 = await Bind("*.*.rabbit", "lazy.#");

    string[] keys =
    {
        "quick.orange.rabbit",  // q1 + q2
        "lazy.orange.elephant", // q1 + q2
        "quick.orange.fox",     // q1 only
        "lazy.brown.fox",       // q2 only
        "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
        "quick.brown.fox",      // nobody → silent drop
    };
    foreach (var key in keys)
        await channel.BasicPublishAsync(exchange, key, body: Encoding.UTF8.GetBytes(key));
    Console.WriteLine($"[x] Published {keys.Length} keys");

    await WaitFor(q1, 3);
    await WaitFor(q2, 4);
    Console.WriteLine($"[q1] matched: {string.Join(", ", q1.Keys.Order())}");
    Console.WriteLine($"[q2] matched: {string.Join(", ", q2.Keys.Order())}");
    Console.WriteLine("[v] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");

    static async Task WaitFor(ConcurrentDictionary<string, byte> set, int n)
    {
        var deadline = DateTime.UtcNow.AddSeconds(30);
        while (set.Count < n && DateTime.UtcNow < deadline) await Task.Delay(50);
        if (set.Count < n) throw new Exception($"expected {n}, saw {set.Count}");
    }
    ```
  </Tab>

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

    EXCHANGE = "topic_logs"

    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

    ch = conn.create_channel
    exchange = ch.topic(EXCHANGE, durable: false)

    bind = lambda do |*patterns|
      sub_ch = conn.create_channel
      queue = sub_ch.queue("", exclusive: true)
      patterns.each { |p| queue.bind(exchange, routing_key: p) }
      seen = []
      mutex = Mutex.new
      queue.subscribe(manual_ack: false, block: false) { |_di, _props, body| mutex.synchronize { seen << body } }
      { seen: seen, mutex: mutex }
    end

    q1 = bind.call("*.orange.*")            # orange animals
    q2 = bind.call("*.*.rabbit", "lazy.#")  # rabbits and everything lazy

    keys = %w[
      quick.orange.rabbit lazy.orange.elephant quick.orange.fox
      lazy.brown.fox lazy.pink.rabbit quick.brown.fox
    ]
    keys.each { |k| exchange.publish(k, routing_key: k) }
    puts " [x] Published #{keys.size} keys"

    wait_for = lambda do |entry, n|
      deadline = Time.now + 30
      loop do
        break if entry[:mutex].synchronize { entry[:seen].uniq.size } >= n || Time.now > deadline

        sleep 0.05
      end
    end

    wait_for.call(q1, 3)
    wait_for.call(q2, 4)
    puts " [q1] matched: #{q1[:seen].uniq.sort}"
    puts " [q2] matched: #{q2[:seen].uniq.sort}"
    puts " [x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped"

    conn.close
    ```
  </Tab>

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

    const EXCHANGE: &str = "topic_logs";

    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 bind(conn: &Connection, patterns: &[&str]) -> Result<lapin::Consumer, Box<dyn std::error::Error>> {
        let ch = conn.create_channel().await?;
        let queue = ch
            .queue_declare("", QueueDeclareOptions { exclusive: true, ..Default::default() }, FieldTable::default())
            .await?;
        for p in patterns {
            ch.queue_bind(queue.name().as_str(), EXCHANGE, p, QueueBindOptions::default(), FieldTable::default())
                .await?;
        }
        Ok(ch
            .basic_consume(queue.name().as_str(), "", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
            .await?)
    }

    async fn collect(label: &str, consumer: &mut lapin::Consumer, want: usize) -> Result<(), Box<dyn std::error::Error>> {
        let mut seen = BTreeSet::new();
        while seen.len() < want {
            let delivery = consumer.next().await.ok_or("consumer closed early")??;
            seen.insert(String::from_utf8_lossy(&delivery.data).into_owned());
        }
        println!("[{label}] matched: {seen:?}");
        Ok(())
    }

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

        let mut q1 = bind(&conn, &["*.orange.*"]).await?;
        let mut q2 = bind(&conn, &["*.*.rabbit", "lazy.#"]).await?;

        let keys = [
            "quick.orange.rabbit",  // q1 + q2
            "lazy.orange.elephant", // q1 + q2
            "quick.orange.fox",     // q1 only
            "lazy.brown.fox",       // q2 only
            "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
            "quick.brown.fox",      // nobody → silent drop
        ];
        for key in keys {
            ch.basic_publish(EXCHANGE, key, BasicPublishOptions::default(), key.as_bytes(), BasicProperties::default())
                .await?;
        }
        println!("[x] Published {} keys", keys.len());

        collect("q1", &mut q1, 3).await?;
        collect("q2", &mut q2, 4).await?;
        println!("[x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");

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

## Single copy on multiple matching bindings [#single-copy-on-multiple-matching-bindings]

If a queue is bound with two patterns that both match a key — like Q2's `*.*.rabbit` and `lazy.#` both matching `lazy.pink.rabbit` — the connector deduplicates the matched-queue set and delivers exactly **one** copy. You never receive a duplicate just because more than one of your own bindings matched.

<Callout type="info">
  **A key matching no binding is dropped.** `quick.brown.fox` matches neither queue, so the connector resolves it to an empty set and writes nothing — no error, no delivery. As with [direct routing](/connectors/rabbitmq/how-to/routing), set `mandatory=true` to receive a `312 NO_ROUTE` for unroutable publishes.
</Callout>

## Related [#related]

<Cards>
  <Card title="Routing (direct)" href="/connectors/rabbitmq/how-to/routing" description="Exact-match routing — the building block that topics generalize with wildcards." />

  <Card title="Pub/Sub (fanout)" href="/connectors/rabbitmq/how-to/pub-sub" description="Broadcast to all subscribers — equivalent to binding everything with #." />

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