下面做一个完整但不花哨的示例：
场景是：
- 订单系统发出“创建包裹运单”任务
- 发货工人消费任务
- 成功就
ack - 临时失败就进入 10 秒重试队列
- 重试多次仍失败，进入死信队列
拓扑图如下：
Producer
|
| routing_key = parcel.create
v
[ parcel.dispatch exchange (direct) ]
|
+--> [ parcel.create.q ]
|
| consumer 处理失败，nack(requeue=False)
v
[ parcel.retry exchange ]
|
+--> [ parcel.create.retry.10s.q ]
| TTL = 10000ms
| 到期后 dead-letter 回主交换机
v
[ parcel.dispatch exchange ]
|
+--> [ parcel.create.q ]
超过最大重试次数：
Consumer 手动转发到
[ parcel.dead exchange ] --> [ parcel.create.dead.q ]#1. 建立拓扑：topology.py
import pika
RABBITMQ_URL = "amqp://guest:guest@localhost:5672/%2F"
MAIN_EXCHANGE = "parcel.dispatch"
RETRY_EXCHANGE = "parcel.retry"
DEAD_EXCHANGE = "parcel.dead"
MAIN_QUEUE = "parcel.create.q"
RETRY_QUEUE = "parcel.create.retry.10s.q"
DEAD_QUEUE = "parcel.create.dead.q"
MAIN_KEY = "parcel.create"
RETRY_KEY = "parcel.create.retry"
DEAD_KEY = "parcel.create.dead"
def main():
params = pika.URLParameters(RABBITMQ_URL)
connection = pika.BlockingConnection(params)
channel = connection.channel()
channel.exchange_declare(exchange=MAIN_EXCHANGE, exchange_type="direct", durable=True)
channel.exchange_declare(exchange=RETRY_EXCHANGE, exchange_type="direct", durable=True)
channel.exchange_declare(exchange=DEAD_EXCHANGE, exchange_type="direct", durable=True)
channel.queue_declare(
queue=MAIN_QUEUE,
durable=True,
arguments={
"x-dead-letter-exchange": RETRY_EXCHANGE,
"x-dead-letter-routing-key": RETRY_KEY,
},
)
channel.queue_declare(
queue=RETRY_QUEUE,
durable=True,
arguments={
"x-message-ttl": 10_000,
"x-dead-letter-exchange": MAIN_EXCHANGE,
"x-dead-letter-routing-key": MAIN_KEY,
},
)
channel.queue_declare(queue=DEAD_QUEUE, durable=True)
channel.queue_bind(queue=MAIN_QUEUE, exchange=MAIN_EXCHANGE, routing_key=MAIN_KEY)
channel.queue_bind(queue=RETRY_QUEUE, exchange=RETRY_EXCHANGE, routing_key=RETRY_KEY)
channel.queue_bind(queue=DEAD_QUEUE, exchange=DEAD_EXCHANGE, routing_key=DEAD_KEY)
print("Topology declared.")
connection.close()
if __name__ == "__main__":
main()#2. 发任务：producer.py
import json
import pika
RABBITMQ_URL = "amqp://guest:guest@localhost:5672/%2F"
MAIN_EXCHANGE = "parcel.dispatch"
MAIN_KEY = "parcel.create"
def main():
params = pika.URLParameters(RABBITMQ_URL)
connection = pika.BlockingConnection(params)
channel = connection.channel()
# 开启 publisher confirms，至少知道 broker 是否接收了发布
channel.confirm_delivery()
payload = {
"order_id": 1001,
"user_id": 88,
"address": "Shanghai Pudong",
"items": [
{"sku": "BOOK-001", "qty": 1},
{"sku": "PEN-009", "qty": 2},
],
}
ok = channel.basic_publish(
exchange=MAIN_EXCHANGE,
routing_key=MAIN_KEY,
body=json.dumps(payload).encode("utf-8"),
properties=pika.BasicProperties(
delivery_mode=2, # 持久化消息
content_type="application/json",
),
mandatory=True,
)
print("Published:", ok)
connection.close()
if __name__ == "__main__":
main()#3. 消费任务：worker.py
import json
import random
import pika
RABBITMQ_URL = "amqp://guest:guest@localhost:5672/%2F"
MAIN_QUEUE = "parcel.create.q"
DEAD_EXCHANGE = "parcel.dead"
DEAD_KEY = "parcel.create.dead"
RETRY_QUEUE = "parcel.create.retry.10s.q"
MAX_RETRY = 3
def count_retries(properties):
headers = properties.headers or {}
x_death = headers.get("x-death", [])
total = 0
for death in x_death:
if death.get("queue") == RETRY_QUEUE:
total += int(death.get("count", 0))
return total
def create_shipping_label(task):
# 用随机数模拟外部物流接口偶发失败
n = random.random()
if n < 0.55:
return True
if n < 0.90:
raise TimeoutError("carrier api timeout")
raise ValueError("invalid address")
def main():
params = pika.URLParameters(RABBITMQ_URL)
connection = pika.BlockingConnection(params)
channel = connection.channel()
# 一次只派一件活给当前工人，避免压太多未确认任务
channel.basic_qos(prefetch_count=1)
def callback(ch, method, properties, body):
task = json.loads(body.decode("utf-8"))
retry_count = count_retries(properties)
print(f"Received order_id={task['order_id']} retry_count={retry_count}")
try:
create_shipping_label(task)
print("Shipping label created.")
ch.basic_ack(delivery_tag=method.delivery_tag)
return
except TimeoutError as e:
# 临时失败：没到上限则走重试链路
if retry_count < MAX_RETRY:
print(f"Temporary failure: {e}. Send to retry.")
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
return
# 超过上限：转死信队列
print(f"Retry exhausted: {e}. Send to dead queue.")
ch.basic_publish(
exchange=DEAD_EXCHANGE,
routing_key=DEAD_KEY,
body=body,
properties=pika.BasicProperties(
delivery_mode=2,
content_type=properties.content_type or "application/json",
headers={
**(properties.headers or {}),
"final_reason": "retry_exhausted",
},
),
)
ch.basic_ack(delivery_tag=method.delivery_tag)
return
except ValueError as e:
# 永久失败：直接进死信
print(f"Permanent failure: {e}. Send to dead queue.")
ch.basic_publish(
exchange=DEAD_EXCHANGE,
routing_key=DEAD_KEY,
body=body,
properties=pika.BasicProperties(
delivery_mode=2,
content_type=properties.content_type or "application/json",
headers={
**(properties.headers or {}),
"final_reason": "invalid_payload",
},
),
)
ch.basic_ack(delivery_tag=method.delivery_tag)
return
channel.basic_consume(queue=MAIN_QUEUE, on_message_callback=callback, auto_ack=False)
print("Worker started.")
channel.start_consuming()
if __name__ == "__main__":
main()这个示例故意把几个最重要的基础动作都放进来了：
exchangequeuebindingrouting key- 手动
ack - 消息持久化
- TTL 重试
- 死信队列
prefetch控制并发领取量
这就是 RabbitMQ 最典型、最实用的一套任务分发心智。