不能靠 RabbitMQ 自身机制防重复，必须在消费者端做幂等控制；需用 direct/topic exchange + 单个共享 queue 实现竞争消费者模式，并通过数据库唯一索引、Redis setex 或状态机校验保障幂等。
直接说结论：不能靠 rabbitmq 自身机制“防重复”，必须在消费者端做幂等控制；多实例共享同一个 queue 是前提，否则连“同一张订单只被一个实例处理”都做不到。
为什么 RabbitMQ 的 fanout/exchange 不能解决这个问题？
fanout exchange 是广播模式，所有绑定的 queue 都会收到完整副本——这适合“不同服务各收一份”，但不适用于“同一服务多个实例只收一份”。如果你让每个实例连自己的 queue，那每条充值订单消息就会被所有实例消费一次，10 个实例就是 10 次重复。
真正该用的是 direct 或 topic exchange + 单个持久化 queue，所有实例共用这个 queue。RabbitMQ 会自动轮询分发消息给空闲消费者（即“竞争消费者”模式），这是防止多实例重复消费的基础设施前提。
- 确保队列声明时
durable=True，避免重启丢配置 - 消费者连接时不要设
exclusive=True，否则只能单实例独占 - 禁用自动 ACK：
auto_ack=False，否则处理中途崩溃会导致消息丢失或重复投递
FastAPI 启动时如何正确声明并绑定 queue？
别在每次请求里连 RabbitMQ，而是在 FastAPI 生命周期钩子中初始化连接和 channel，并声明 queue 与 exchange 绑定关系。推荐用 lifespan（Python 3.9+）或 startup event。
关键点：
RabbitMQ 4.2.3 是 2026 年初发布的重要稳定更新版本，重点修复了 Khepri 元数据存储相关问题，并改进了监控性能。对于使用 Docker、Kubernetes 或微服务架构的开发团队来说，该版本兼容性和稳定性表现较好。
- exchange 名、routing key、queue 名必须全项目统一，且硬编码或从配置读取，不能动态生成
- 声明 queue 时加
arguments={"x-dead-letter-exchange": "dlx.exchange"}，为异常消息留退路 - 绑定必须显式调用
channel.queue_bind()，仅声明 queue 不等于完成绑定
示例片段（使用 aio-pika）：
async def lifespan(app: FastAPI):
connection = await connect_robust("amqp://guest:guest@localhost/")
channel = await connection.channel()
await channel.set_qos(prefetch_count=1) # 关键：限制未确认消息数，防堆积
exchange = await channel.declare_exchange("recharge.exchange", type="direct")
queue = await channel.declare_queue("recharge.queue", durable=True)
await queue.bind(exchange, routing_key="recharge.create")
app.state.rabbitmq_channel = channel
yield
await connection.close()
充值订单幂等性的三种落地方式，按优先级排序
消息体里必须带可唯一标识订单的字段，比如 order_id 或业务组合键（如 "recharge_" + user_id + "_" + timestamp）。没这个 ID，一切幂等都是空谈。
-
首选数据库唯一索引：建一张
recharge_processed表，主键或唯一索引为order_id。消费逻辑开头就INSERT INTO recharge_processed (order_id) VALUES (...)，失败则说明已处理，直接 return；成功则继续扣款、发通知等。简单、强一致、无需额外组件。 -
Redis setnx + 过期时间：用
redis.setex("recharge:processed:" + order_id, 3600, "1")。注意不是setnx单独用，要带过期（防 Redis 故障导致锁永久残留）。适合高并发场景，但依赖 Redis 可用性。 -
状态机校验：查订单当前状态，比如
SELECT status FROM recharge_orders WHERE order_id = ?。若已是"success"或"processing"，直接跳过；否则更新为"processing"并执行。需配合 SELECT FOR UPDATE 或乐观锁（version字段），否则并发下仍可能重复。
最容易被忽略的两个细节
一是消息体里的 order_id 必须由生产者生成并透传，不能在消费者端拼接或推导——否则不同实例可能算出不同 ID；二是手动 ACK 的位置必须在幂等校验通过且核心业务执行完毕之后，不能在 DB 插入前就 await delivery.message.ack()，否则幂等失败时消息已确认，再也不会重试。
如果充值涉及第三方支付回调，还要额外检查回调签名和幂等 ID 是否匹配，这部分不在 RabbitMQ 范畴，但实际项目中常在这里翻车。
