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 by pattern matching on routing keys using wildcards, enabling flexible subscriptions for selective message delivery.

Wildcard Rules

• * matches exactly one dot-separated word
• # matches zero or more words

Topic Exchange Flow

flowchart LR
    P[Producer] --> T[Topic Exchange]
    T -->|"error.#"| Q1[All Errors]
    T -->|"#.auth.#"| Q2[Auth Events]
    T -->|"critical.*"| Q3[Critical Alerts]
    style T fill:#f90,color:#fff

import pika
import json

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

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

channel.queue_declare(queue='all_errors', durable=True)
channel.queue_declare(queue='auth_events', durable=True)
channel.queue_declare(queue='critical_alerts', durable=True)

channel.queue_bind(exchange='topic_logs', queue='all_errors', routing_key='error.#')
channel.queue_bind(exchange='topic_logs', queue='auth_events', routing_key='#.auth.#')
channel.queue_bind(exchange='topic_logs', queue='critical_alerts', routing_key='critical.*')

events = [
    ('error.auth.login', 'Failed login for alice'),
    ('error.db.timeout', 'Database connection timeout'),
    ('info.auth.logout', 'User bob logged out'),
    ('warning.cache.memory', 'Cache memory high'),
    ('critical.disk', 'Disk space critical'),
]

for key, msg in events:
    channel.basic_publish(exchange='topic_logs', routing_key=key, body=msg)
    print(f"Published: {key}")

connection.close()

Expected output:

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

# All Errors consumer
import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.exchange_declare(exchange='topic_logs', exchange_type='topic', durable=True)
ch.queue_declare(queue='all_errors', durable=True)
ch.queue_bind(exchange='topic_logs', queue='all_errors', routing_key='error.#')

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

ch.basic_consume(queue='all_errors', on_message_callback=cb, auto_ack=True)
ch.start_consuming()

Expected output:

[error.auth.login] Failed login for alice
[error.db.timeout] Database connection timeout

Designing Routing Key Hierarchies

<domain>.<action>.<status>
<severity>.<component>.<subcomponent>

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

Use 3-4 levels. Consistent conventions across all services make subscriptions predictable.

Common Mistakes

1. Confusing * with #

error.* matches error.db but not error.db.timeout (two more words). error.# matches both. Test patterns with sample routing keys.

2. Binding with # When Fanout is Better

If you bind a queue with #, it receives all messages — same as fanout. Use fanout explicitly for broadcasts.

3. Inconsistent Key Hierarchies

If one team uses error.db.timeout and another uses db/error/timeout, subscriptions break. Document and enforce a routing key convention.

4. Too Many Levels

Routing keys with 8+ levels are hard to read and pattern-match. Keep it to 3-4 levels maximum.

5. Case Sensitivity

Error.Auth.Login is different from error.auth.login. Use lowercase consistently.

Practice Questions

1. What is the difference between * and #?

* matches exactly one word. # matches zero or more words. a.*.b matches a.x.b but not a.x.y.b. a.#.b matches both.

2. What does # match?

# matches zero or more dot-separated words. Binding with # alone matches all routing keys.

3. How do you subscribe to all events from the auth component?

Use #.auth.# or *.auth.* depending on your hierarchy. #.auth.# catches any routing key with auth anywhere.

4. Can I bind a queue with multiple patterns?

Yes. A queue can have multiple bindings to the same or different exchanges. Messages matching any pattern are delivered.

Challenge

Design a topic exchange hierarchy for a multi-tenant SaaS platform. Events include user actions, billing events, system health, and security alerts. Each tenant should be able to subscribe to their own events.

FAQ

Mini Project: Event Router

import pika, json, threading, time

def start_subscriber(name, patterns):
    conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    ch = conn.channel()
    ch.exchange_declare(exchange='app_events', exchange_type='topic', durable=True)
    q = ch.queue_declare(queue=name, durable=True)
    for p in patterns:
        ch.queue_bind(exchange='app_events', queue=q.method.queue, routing_key=p)

    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)
    ch.start_consuming()

subs = [
    ('error_collector', ['error.#']),
    ('auth_monitor', ['#.auth.#']),
    ('payment_handler', ['#.payment.#']),
]
for name, pats in subs:
    t = threading.Thread(target=start_subscriber, args=(name, pats), daemon=True)
    t.start()

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

events = [
    ('error.auth.login', 'Failed login'),
    ('info.payment.completed', 'Payment received'),
    ('error.payment.declined', 'Card declined'),
]
for key, msg in events:
    ch.basic_publish(exchange='app_events', routing_key=key, body=msg)

time.sleep(1)

Expected output:

[error_collector] (error.auth.login) Failed login
[auth_monitor] (error.auth.login) Failed login
[error_collector] (error.payment.declined) Card declined
[payment_handler] (info.payment.completed) Payment received
[payment_handler] (error.payment.declined) Card declined

What's Next

Now that you understand topic exchange, explore headers exchange for attribute-based routing, then learn about dead letter exchange for handling failed messages.