How to Simulate RabbitMQ Exchange Routing in Python

Simulate RabbitMQ exchange routing using a nested dict, matching routing keys against patterns like error.* and info.# to return bound queues.

Medium Python 3.9+ Aug 9, 2026 Streaming & messaging 14 views 0 copies

rabbitmq routing messaging simulation exchanges

Python code

46 lines

Python 3.9+

from collections import defaultdict

def route_message(exchanges, exchange_name, routing_key):
    """
    Simulate RabbitMQ exchange routing using a nested dict structure.
    Returns list of queue names that match the routing key.
    """
    queues = exchanges.get(exchange_name, {})
    matched = []
    
    for pattern, queue_list in queues.items():
        if routing_key == pattern:
            matched.extend(queue_list)
        elif pattern.endswith("*") and routing_key.startswith(pattern[:-1]):
            matched.extend(queue_list)
        elif pattern.endswith("#") and routing_key.startswith(pattern[:-1]):
            matched.extend(queue_list)
    
    return matched

if __name__ == "__main__":
    # Simulated exchange bindings: exchange -> {routing_pattern: [queues]}
    exchanges = {
        "logs": {
            "error.*": ["error_queue", "all_logs"],
            "warning.*": ["warning_queue", "all_logs"],
            "info.#": ["info_queue", "all_logs"]
        },
        "orders": {
            "created": ["order_created_worker"],
            "fulfilled": ["order_fulfilled_worker"]
        }
    }

    # Test various routing keys
    test_cases = [
        ("logs", "error.database"),
        ("logs", "info.user.login"),
        ("orders", "created"),
        ("orders", "unknown.key"),
        ("logs", "warning.timeout"),
    ]

    for exchange, routing_key in test_cases:
        matched_queues = route_message(exchanges, exchange, routing_key)
        print(f"Exchange={exchange}, RoutingKey={routing_key} -> Queues={matched_queues}")

Output

stdout

Exchange=logs, RoutingKey=error.database -> Queues=['error_queue', 'all_logs']
Exchange=logs, RoutingKey=info.user.login -> Queues=['info_queue', 'all_logs']
Exchange=orders, RoutingKey=created -> Queues=['order_created_worker']
Exchange=orders, RoutingKey=unknown.key -> Queues=[]
Exchange=logs, RoutingKey=warning.timeout -> Queues=['warning_queue', 'all_logs']

How it works

The route_message function looks up the exchange's binding dict and iterates each pattern. It first tries an exact match, then treats * as a wildcard for a single segment and # for multiple segments by checking if the routing key starts with the prefix before the wildcard. Because patterns are stored per exchange in a nested dict, the function stays fast and readable. This mirrors how RabbitMQ matches direct, topic, and fanout exchanges in a lightweight, dependency-free way.

Common mistakes

Using `pattern.endswith("#")` to match any suffix but forgetting that `#` can also match zero segments.
Not handling case where the exchange name doesn't exist — the code defaults to an empty dict, which is safe.
Assuming `*` matches only one segment; in real RabbitMQ, `*` does not cross dots, but this simple version treats it as a prefix.
Forgetting that a routing key may match multiple patterns and queues — the function returns duplicates if a queue is bound to multiple keys.

Variations

Use `fnmatch` for more flexible wildcard matching patterns.
Implement a class-based Exchange with binding methods for a more production-like design.

Real-world use cases

Testing message routing logic locally without spinning up a RabbitMQ broker.
Building a lightweight event bus abstraction for unit tests in microservice development.
Prototyping topic exchange patterns before implementing them in production infrastructure.

How to Simulate RabbitMQ Exchange Routing in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes