RabbitMQ: Flexible Message Routing

Producer                  Exchange（路由器）          Queue                 Consumer
  │                          │                        │                       │
  └─ publish(routing_key) ──┬─ direct ────────────────┐                       │
                            ├─ fanout ────────────────┼───────────────────────┤
                            ├─ topic ─────────────────┘                       │
                            └─ headers                          bind ────────┘

Exchange 四种类型：
  direct   routing_key 精确匹配
  fanout   广播给所有绑定队列（忽略 key）
  topic    模式匹配：* 一词，# 多词（orders.* / orders.# ）
  headers  按消息头匹配（少用）

队列是真正存消息的地方；交换器不存消息

# 声明（Go / Python / Java 客户端 API 同构）
# exchange.declare(name, type='topic', durable=True)
# queue.declare(name='orders.create', durable=True,
#               arguments={
#                 'x-dead-letter-exchange': 'orders.dlx',
#                 'x-message-ttl': 60000,
#                 'x-max-priority': 10
#               })
# queue.bind(queue, exchange, routing_key='orders.create')
# basic.publish(exchange, routing_key, body, mandatory=True,
#               properties=Persistent + MessageId + ContentTypeJSON)

消息可靠性 = 发布者确认 + 消费者手动 ack + 持久化

发布者确认（publisher confirm）：
 Broker 收到并落盘后回 confirm；未回的发布可重发
 配合 mandatory=true：无队列匹配时触发 return，防丢

消费者手动 ack：
 basic.consume(auto_ack=False)
 处理成功 → basic.ack
 处理失败 → basic.nack(requeue=False) → 死信
 消费者掉线（连接断）→ 消息自动 requeue 给其他消费者

死信队列 DLX：消息在以下情况转入死信交换器：
  1. 被 nack/reject 且 requeue=false
  2. TTL 到期
  3. 队列达到最大长度

持久化三件套：
  exchange durable + queue durable + message persistent
三者缺一，重启即丢消息

Auto ack is the default in many tutorials and the cause of silent message loss: if the consumer crashes mid-processing the message is gone. Always use manual ack and idempotent consumers.

basic.qos(prefetch_count=N)   每个消费者最多同时持有 N 条未 ack

prefetch=1：逐条处理，吞吐低但绝不堆积
prefetch 过大：客户端内存涨，宕机重投多
经验值：处理耗时 10ms 级取 20-50，秒级取个位数

Broker 侧流控（内存/磁盘水位）：
  水位线触发 → 阻塞发布者（对资源警告），不丢消息
  disk_free_limit：磁盘低于阈值直接拒绝写入

监控三件事：队列深度、未 ack 数、消费者利用率

Producer                  Exchange（路由器）          Queue                 Consumer
  │                          │                        │                       │
  └─ publish(routing_key) ──┬─ direct ────────────────┐                       │
                            ├─ fanout ────────────────┼───────────────────────┤
                            ├─ topic ─────────────────┘                       │
                            └─ headers                          bind ────────┘

Exchange 四种类型：
  direct   routing_key 精确匹配
  fanout   广播给所有绑定队列（忽略 key）
  topic    模式匹配：* 一词，# 多词（orders.* / orders.# ）
  headers  按消息头匹配（少用）

队列是真正存消息的地方；交换器不存消息

# 声明（Go / Python / Java 客户端 API 同构）
# exchange.declare(name, type='topic', durable=True)
# queue.declare(name='orders.create', durable=True,
#               arguments={
#                 'x-dead-letter-exchange': 'orders.dlx',
#                 'x-message-ttl': 60000,
#                 'x-max-priority': 10
#               })
# queue.bind(queue, exchange, routing_key='orders.create')
# basic.publish(exchange, routing_key, body, mandatory=True,
#               properties=Persistent + MessageId + ContentTypeJSON)

消息可靠性 = 发布者确认 + 消费者手动 ack + 持久化

发布者确认（publisher confirm）：
 Broker 收到并落盘后回 confirm；未回的发布可重发
 配合 mandatory=true：无队列匹配时触发 return，防丢

消费者手动 ack：
 basic.consume(auto_ack=False)
 处理成功 → basic.ack
 处理失败 → basic.nack(requeue=False) → 死信
 消费者掉线（连接断）→ 消息自动 requeue 给其他消费者

死信队列 DLX：消息在以下情况转入死信交换器：
  1. 被 nack/reject 且 requeue=false
  2. TTL 到期
  3. 队列达到最大长度

持久化三件套：
  exchange durable + queue durable + message persistent
三者缺一，重启即丢消息

很多教程默认自动 ack，是消息静默丢失的元凶：处理到一半消费者挂了，消息就没了。一定要手动 ack，并把消费者做成幂等。

basic.qos(prefetch_count=N)   每个消费者最多同时持有 N 条未 ack

prefetch=1：逐条处理，吞吐低但绝不堆积
prefetch 过大：客户端内存涨，宕机重投多
经验值：处理耗时 10ms 级取 20-50，秒级取个位数

Broker 侧流控（内存/磁盘水位）：
  水位线触发 → 阻塞发布者（对资源警告），不丢消息
  disk_free_limit：磁盘低于阈值直接拒绝写入

监控三件事：队列深度、未 ack 数、消费者利用率