TopolearnChoosing the exchange type is how you decide who receives which messages without changing producers.
Producers publish to an exchange with a routing key; the exchange type decides which bound queues get a copy:
- direct: delivers to queues whose binding key equals the routing key. Good for routing by type or severity:
error,warning. - fanout: ignores the routing key and delivers to every bound queue. Classic publish/subscribe: broadcast cache invalidations or events.
- topic: matches the routing key against binding patterns. Keys are dot-separated words like
order.eu.created; in patterns*matches exactly one word and#matches zero or more.order.*.createdororder.#. - headers: routes on message header values instead of the key, with
x-match: allorany. Rarely needed.
If no binding matches, the message is dropped silently, unless it's published with the mandatory flag (it's returned to the publisher) or the exchange has an alternate exchange configured.
A queue can have many bindings, and an exchange can be bound to another exchange. Exchanges are declared like queues, idempotently, with exchange_declare.
Typical design: one topic exchange per domain (orders, payments) with descriptive routing keys, and each consuming service declaring and binding its own queue for the keys it cares about. Producers then never need to know who consumes.
Example
import pika
ch = pika.BlockingConnection(pika.ConnectionParameters("localhost")).channel()
ch.confirm_delivery() # each publish waits until routed
ch.exchange_declare(exchange="logs", exchange_type="direct")
ch.exchange_declare(exchange="broadcast", exchange_type="fanout")
ch.queue_declare(queue="errors")
ch.queue_bind(queue="errors", exchange="logs", routing_key="error")
for q in ("cache-a", "cache-b"):
ch.queue_declare(queue=q)
ch.queue_bind(queue=q, exchange="broadcast") # key ignored
ch.basic_publish(exchange="logs", routing_key="error", body=b"disk full")
ch.basic_publish(exchange="logs", routing_key="info", body=b"started") # dropped
ch.basic_publish(exchange="broadcast", routing_key="", body=b"invalidate")
for q in ("errors", "cache-a", "cache-b"):
print(q, ch.queue_declare(queue=q, passive=True).method.message_count)
# errors 1 cache-a 1 cache-b 1Remember
- direct = exact key, fanout = everyone, topic = patterns with * and #
- Unmatched messages are dropped unless mandatory or an alternate exchange is used
- Consumers own their queues and bindings; producers only know the exchange
Watch out for
- Assuming RabbitMQ keeps unroutable messages; with no matching binding they vanish.