用 RabbitMQ 构建可靠消息系统时：如何保证不丢消息…

设计、实现和运维 RabbitMQ 消息队列系统时，覆盖可靠性、HA、安全、性能与监控的完整实践清单。

适用场景

解决思路

1. 先写失败测试：用 pytest + unittest.mock 写单元测试验证 ack/nack、prefetch、publisher confirms、DLX 路由，再用真实 RabbitMQ 写集成测试（连接失败时 pytest.skip）。
2. 实现消费者时使用手动确认：auto_ack=False，处理成功调用 ch.basic_ack(delivery_tag=method.delivery_tag)，JSON 解析失败或处理异常调用 ch.basic_nack(delivery_tag=..., requeue=False) 送入 DLX。
3. 声明队列和交换机时设置 durable=True，发布消息时设置 properties=pika.BasicProperties(delivery_mode=2) 使消息持久化。
4. 启用 publisher confirms：channel.confirm_delivery()，配合 mandatory=True 检测不可路由消息，捕获 UnroutableError 和 NackError。
5. 配置 DLX：声明 fanout 类型的 dlx 交换机与失败队列并绑定，主队列通过 arguments 设置 x-dead-letter-exchange、x-message-ttl、x-max-length、x-max-retries。
6. 根据处理耗时调优 prefetch：快消费者（<100ms）设 20-50，中等（100ms-1s）设 5-20，慢消费者（>1s）设 1-5，通过 channel.basic_qos(prefetch_count=N) 设置，并监控 consumer utilization 调整。
7. 批量发布时先 confirm_delivery()，循环 basic_publish 不逐条等待，最后用 channel.get_waiting_message_count() 强制刷新确认，捕获 NackError 处理被拒消息。
8. 使用连接池复用连接：用 queue.Queue 维护连接池，get_channel 取连接、return_connection 归还，连接失效时替换新连接，避免每次操作新建 BlockingConnection。
9. 大积压场景使用 lazy 队列（x-queue-mode: lazy）或 quorum 队列（x-queue-type: quorum 配合 x-max-in-memory-length 限制内存中消息数）。
10. 高可用使用 quorum 队列替代 classic mirrored 队列：x-queue-type: quorum，配合 x-delivery-limit 做毒消息检测，集群节点数用奇数（3/5/7）。
11. 异步高吞吐场景使用 SelectConnection + confirm_delivery(on_confirm) 回调，避免同步 confirm 阻塞每条消息。
12. 序列化优化：结构化数据用 msgpack 并设置 content_type='application/msgpack'，二进制数据直接传 bytes 并设置 content_type='application/octet-stream'，避免 JSON+base64。
13. Topic 交换机路由：用 * 匹配一个单词、# 匹配零或多个单词，如 *.error、db.*、user.#，为不同队列绑定不同 routing key 模式。
14. 生产配置 rabbitmq.conf：启用 TLS（listeners.ssl.default=5671、verify_peer、fail_if_no_peer_cert=true），设置 vm_memory_high_watermark.relative=0.5、disk_free_limit.absolute=10GB、cluster_partition_handling=autoheal、heartbeat=60、channel_max=2048。
15. 安全加固：rabbitmqctl delete_user guest 删除默认用户，创建 admin 和受限 app_user，用 vhost 隔离环境，用 set_topic_permissions 限制交换机级发布权限。
16. 客户端 TLS 连接：用 ssl.create_default_context(cafile=...) 并设置 check_hostname=True、verify_mode=CERT_REQUIRED，通过 pika.SSLOptions 传入 ConnectionParameters（port=5671）。
17. 凭据用 secrets 管理（如 Kubernetes Secret 挂载环境变量），绝不硬编码或提交到版本库。
18. 实现连接重试：捕获 AMQPConnectionError，用指数退避（time.sleep(2 ** retries)）重试最多 5 次。
19. 监控与告警：暴露 Prometheus 指标，监控队列深度（如 rabbitmq_queue_messages{queue="tasks"} > 10000）、消息速率、consumer utilization、内存/磁盘告警、集群分区事件，用 Grafana 做仪表盘。
20. 提交前检查：单元与集成测试通过、TLS 启用、guest 禁用、强认证、vhost 权限、内存磁盘阈值、Prometheus 监控、告警、关键队列持久化、集群分区处理、备份恢复文档、日志聚合、性能基准达标。
21. 性能测试：基准发布 10000 条 1KB 消息测吞吐（目标 >1000 msg/s），测消费延迟（平均 <10ms、P99 用 statistics.quantiles 计算）。
22. 用 pytest 标记区分测试：@pytest.mark.integration、slow、performance，用 -m "not integration" 跑快速单元测试。
23. 避免常见错误：不用 auto_ack=True、不用非持久化队列、不设无限 prefetch、不无限 requeue（用 DLX）、不用 classic mirrored 队列、不忽略连接失败、不忽略队列深度。

适用边界 · 注意事项

• 不使用 RabbitMQ 而用 Kafka、SQS、Redis 等其它消息系统时
• 仅做一次性脚本或本地原型、不需要持久化/HA/安全加固时
• 团队没有 RabbitMQ 运维能力且无法承担集群、TLS、监控的维护成本时