Topic Exchange — Complete Guide

In this tutorial, you will learn about Topic Exchange. We cover key concepts, practical examples, and best practices to help you master this topic.

A topic exchange routes messages to queues based on routing key pattern matching, enabling flexible subscriptions with wildcards for selective message delivery.

What You'll Learn

By the end of this lesson, you will understand how topic exchanges use routing key patterns, the wildcard characters * and #, how to implement selective subscriptions, and when to use topic exchanges over direct or fanout.

Why It Matters

In real systems, not every consumer needs every message. A logging service needs all error messages. An analytics service needs only payment events. Topic exchanges let each consumer specify exactly which messages they want using simple pattern matching.

Real-World Use

A system monitoring platform categorizes events by severity and component: error.database.timeout, warning.cache.high-memory, info.auth.login. The on-call team subscribes to error.# to see all errors. The database team subscribes to #.database.# for database-related events.

Topic Exchange Flow

flowchart LR
    P[Publisher] --> T[Topic Exchange]
    T -->|"Binding: error.#"| Q1[All Errors Queue]
    T -->|"Binding: #.auth.#"| Q2[Auth Events Queue]
    T -->|"Binding: critical.*"| Q3[Critical Alerts]
    Q1 --> C1[Error Consumer]
    Q2 --> C2[Auth Consumer]
    Q3 --> C3[Critical Alert Consumer]
    style T fill:#f90,color:#fff

Wildcard Rules

• * matches exactly one word (a dot-separated segment)
• # matches zero or more words

Examples:

• error.# matches error, error.db, error.db.timeout
• *.error matches db.error, auth.error, but not db.error.timeout
• #.auth.# matches auth, error.auth, info.auth.login

Implementing Topic Exchange

import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='system_events', exchange_type='topic', durable=True)

events = [
    ('error.auth.login', {'user': 'alice', 'reason': 'invalid_password'}),
    ('error.db.timeout', {'query': 'SELECT * FROM orders'}),
    ('info.auth.logout', {'user': 'bob'}),
    ('warning.cache.memory', {'usage': 85}),
    ('critical.disk', {'usage': 97, 'device': '/dev/sda1'}),
]

for routing_key, data in events:
    channel.basic_publish(
        exchange='system_events',
        routing_key=routing_key,
        body=json.dumps(data),
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f"Published: {routing_key}")

connection.close()

Expected output:

Published: error.auth.login
Published: error.db.timeout
Published: info.auth.logout
Published: warning.cache.memory
Published: critical.disk

# Consumer with pattern subscription
import pika, json

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.exchange_declare(exchange='system_events', exchange_type='topic', durable=True)

q = ch.queue_declare(queue='', exclusive=True)

binding_keys = ['error.#', 'critical.*']
for bk in binding_keys:
    ch.queue_bind(exchange='system_events', queue=q.method.queue, routing_key=bk)
    print(f"Bound with: {bk}")

def cb(c, m, p, body):
    print(f"[{m.routing_key}] {body.decode()}")

ch.basic_consume(queue=q.method.queue, on_message_callback=cb, auto_ack=True)
ch.start_consuming()

Expected output:

Bound with: error.#
Bound with: critical.*
[error.auth.login] {"user": "alice", "reason": "invalid_password"}
[error.db.timeout] {"query": "SELECT * FROM orders"}
[critical.disk] {"usage": 97, "device": "/dev/sda1"}

The consumer subscribes to error.# (all errors) and critical.* (critical alerts at one level). It does not receive info.auth.logout or warning.cache.memory.

Designing Routing Key Hierarchies

A well-designed routing key hierarchy makes subscriptions powerful:

<severity>.<component>.<subcomponent>.<action>

Examples: error.payment.gateway.timeout, info.user.account.created, warning.storage.disk.low

Use three to four levels. Too many levels make routing keys verbose. Too few limit subscription granularity.

Common Mistakes

1. Confusing * with #

* matches exactly one word. # matches zero or more. error.* matches error.db but not error.db.timeout. error.# matches both. Choose carefully.

2. Using Vague Routing Keys

Routing keys like event1, event2 defeat the purpose of topic exchanges. Design a consistent hierarchy: <domain>.<action>.<status>. Document the convention so all teams follow it.

3. Binding Too Broadly

Binding with # makes the topic exchange behave like a fanout exchange. If you need all messages, use fanout explicitly. Topic exchanges shine with selective bindings.

4. Not Documenting Routing Key Conventions

In a microservice architecture, different teams produce and consume events. Without documented routing key conventions, consumers bind to patterns that never match.

5. Mixing Case and Underscores

Routing keys are case-sensitive and should use lowercase with dots as separators. Error.Auth and error_auth are different from error.auth. Establish a standard.

Practice Questions

1. How does a topic exchange differ from direct?

Direct exchange routes by exact routing key match. Topic exchange routes by pattern match using * and # wildcards. Topic is more flexible for selective subscriptions.

2. What does #.error match?

It matches any routing key that ends with error. Examples: db.error, auth.service.error. The # matches zero or more words before error.

3. How do you implement a topic exchange consumer that gets all events except debug?

Bind with multiple patterns that exclude debug events, or use a negative approach: bind with # and filter in the consumer. RabbitMQ does not support negative bindings.

4. What routing key pattern would subscribe to all database errors?

#.database.error or error.database.# depending on the hierarchy design. Both match database errors at different positions in the hierarchy.

Challenge

Design a routing key hierarchy for a multi-service platform: services include auth, payment, inventory, shipping, and notification. Each has create, update, delete, and error events. Design routing keys and subscriptions for the operations team (all errors), the auth team (auth events), and the finance team (payment events).

FAQ

Mini Project: Log Router with Topic Exchange

import pika
import json
import threading
import time

def start_log_collector(name, patterns, exchange):
    conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    ch = conn.channel()
    ch.exchange_declare(exchange=exchange, exchange_type='topic', durable=True)
    q = ch.queue_declare(queue=name, durable=True)

    for pattern in patterns:
        ch.queue_bind(exchange=exchange, queue=q.method.queue, routing_key=pattern)

    def cb(c, m, p, body):
        print(f"[{name}] ({m.routing_key}) {body.decode()}")

    ch.basic_consume(queue=q.method.queue, on_message_callback=cb, auto_ack=True)
    print(f"[{name}] Bound: {patterns}")
    ch.start_consuming()

threads = []
collectors = [
    ('all_errors', ['error.#']),
    ('auth_events', ['#.auth.#']),
    ('critical_alerts', ['critical.#']),
]
for name, patterns in collectors:
    t = threading.Thread(target=start_log_collector, args=(name, patterns, 'logs'), daemon=True)
    t.start()
    threads.append(t)

time.sleep(1)
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.exchange_declare(exchange='logs', exchange_type='topic', durable=True)

logs = [
    ('error.auth.login', 'Invalid password for alice'),
    ('error.db.connection', 'Connection pool exhausted'),
    ('info.auth.logout', 'User bob logged out'),
    ('critical.disk.full', '/dev/sda1 is 98% full'),
]
for key, msg in logs:
    ch.basic_publish(exchange='logs', routing_key=key, body=msg)

time.sleep(1)

Expected output:

[all_errors] Bound: ['error.#']
[auth_events] Bound: ['#.auth.#']
[critical_alerts] Bound: ['critical.#']
[all_errors] (error.auth.login) Invalid password for alice
[auth_events] (error.auth.login) Invalid password for alice
[all_errors] (error.db.connection) Connection pool exhausted
[critical_alerts] (critical.disk.full) /dev/sda1 is 98% full

What's Next

Now that you understand topic exchanges, explore headers exchanges for attribute-based routing, then dive into message tracing for debugging distributed message flows.