开发 RabbitMQ/AMQP 消息队列应用时…

适用于基于 RabbitMQ 与 AMQP 构建发布/订阅、任务分发、RPC、事件分发等消息队列应用的开发与运维。

适用场景

解决思路

1. 按 RabbitMQ broker 模型设计：publisher 发送消息，exchange 按 binding 路由到 queue，consumer 消费 queue 中的消息。
2. 使用 AMQP 0-9-1 作为常用应用层二进制协议与 RabbitMQ 交互。
3. 设计时以可靠性、可扩展性和容错为目标，并确保实现中不留下 TODO、占位符或缺失部分。
4. 选择 Direct Exchange 做单播路由：按 routing key 精确匹配。
5. 选择 Fanout Exchange 做广播：忽略 routing key，路由到所有绑定队列。
6. 选择 Topic Exchange 做多播路由：按通配符模式匹配 routing key。
7. 选择 Headers Exchange 做复杂路由：按消息 header 属性匹配。
8. 为需要 broker 重启后仍存在的队列声明 durable 队列；仅内存场景使用 transient 队列。
9. 选择队列持久性时考虑元数据存储：durable 队列元数据存磁盘，transient 队列元数据尽量存内存。
10. 通过 bindings 把 exchange 连接到 queue，并设置 binding key/规则决定哪些消息进入哪些队列。
11. 必要时为同一 exchange 配置多个 binding，连接到多个队列。
12. 保持队列规模较小，避免大队列造成高 RAM 压力。
13. 在 RAM 受限时预期 RabbitMQ 会把消息刷盘并影响性能，因此要控制队列积压。
14. 监控 queue depth，并实现 backpressure 机制。
15. 使用 TTL 自动删除过期消息。
16. 普通场景可选 Classic Queues，它支持 lazy queues、priorities 等全部特性。
17. 在 RabbitMQ 3.8+ 中，需要高可用和数据安全时使用 Quorum Queues，并用 `x-queue-type: quorum` 声明。
18. 对需要持久化和容错的队列使用 quorum 队列；注意它不能通过 policy 设置，必须由客户端声明。
19. 不需要持久化时使用 transient messages 获得最快吞吐。
20. 使用 persistent messages 时要注意消息会立即写盘并影响吞吐。
21. 理解单个队列是单线程的，一个队列约处理到 50,000 条消息，通过多个队列提升多核吞吐。
22. 按底层节点核数创建相应数量的队列。
23. 在应用中实现 connection pooling，复用连接。
24. 控制连接数量：每个连接约 100 KB RAM，TLS 下更高，数千连接会显著加重 broker 负担。
25. 使用 channel 在单个 TCP 连接上多路复用，发布和消费都通过 channel 进行。
26. 按线程/操作创建 channel，而不是按消息创建 channel。
27. 不再需要 channel 时关闭它以释放资源。
28. 设置 prefetch(QoS) 控制消费者一次接收的未确认消息数量，以优化吞吐。
29. 慢消费者或需要公平分发时使用低 prefetch（1-10）。
30. 追求吞吐时可用高 prefetch（100+），但要接受分发不均风险。
31. 从适中值（50-100）开始，并根据指标调优 prefetch。
32. 优先考虑手动 ack 模式作为默认规则，以获得更好可靠性。
33. 需要更高吞吐且可接受较弱失败保证时使用 auto-ack。
34. 处理完成后立即 ack，释放服务端资源。
35. 持久化消息设置 `delivery_mode=2`，使其在 broker 重启后存活。
36. 用 `content_type` 标明 payload 格式。
37. 在请求/响应模式中包含 `correlation_id` 匹配请求和响应。
38. 对时效性消息设置 `expiration`。
39. 发布时使用 publisher confirms 保证可靠发布。
40. 正确处理 mandatory flag 返回的不可路由消息。
41. 尽可能批量发布消息以提升吞吐。
42. 控制消息大小，避免大消息影响性能。
43. 配置 Dead Letter Exchange(DLX) 处理失败消息。
44. 在消息被 requeue=false 拒绝、TTL 过期、队列长度超限时，将其路由到 DLX。
45. 单独处理死信消息，用于分析和重试。
46. 不要将失败消息无限 requeue 到同一队列，避免自造 DoS 重试循环。
47. 实现带递增延迟的 retry queues。
48. 将问题消息转发到 timeout 队列。
49. 重试延迟示例：1s、5s、30s，然后进入死信。
50. 对下游故障实现 circuit breaker 模式。
51. 当消费者无法处理消息时，防止队列持续积压。
52. 重连尝试使用 exponential backoff。
53. 为高可用将 RabbitMQ 部署为集群。
54. 对需要复制的持久队列使用 quorum queues，并配置合适的副本数。
55. Classic Queues 镜像使用 `ha-mode` policy。
56. 新部署优先使用 quorum queues，而不是 mirrored classic queues。
57. 使用强密码并定期轮换。
58. 企业部署可考虑集成 LDAP 或 OAuth2。
59. 启用 TLS 加密连接。
60. 为每个应用/服务使用独立凭据。
61. 多租户场景使用 vhost-based isolation。
62. 只授予最小必要权限（configure、write、read）。
63. 使用 permission patterns 限制对特定资源的访问。
64. 监控 queue depth，包括 messages ready 和 messages unacknowledged。
65. 监控消息速率，包括 publish、deliver、acknowledge。
66. 监控连接数和 channel 数。
67. 监控内存和磁盘使用。
68. 监控 consumer utilization。
69. 启用 management plugin 获得监控 UI。
70. 使用 HTTP API 做程序化监控。
71. 将指标导出到 Prometheus/Grafana。
72. 对 queue depth 超过阈值设置告警。
73. 监控内存和磁盘 alarm。
74. 跟踪消费者断开连接。
75. 关注消息速率异常。
76. 单元测试中 mock RabbitMQ client 做隔离测试。
77. 测试消息序列化/反序列化。
78. 验证 exchange 和 queue 声明。
79. 集成测试使用容器化 RabbitMQ。
80. 测试失败场景，包括连接丢失、broker 重启。
81. 验证消息 ack 行为。
82. 用真实消息速率做负载测试。
83. Work Queues 模式：单个队列上多个消费者做负载分发。
84. Work Queues 模式：使用 prefetch 确保公平分发。
85. Work Queues 模式：任务完成后 ack。
86. Publish/Subscribe 模式：使用 fanout exchange 广播。
87. Publish/Subscribe 模式：每个订阅者拥有自己的队列并绑定到 exchange。
88. Publish/Subscribe 模式：需要过滤订阅时考虑 topic exchange。
89. RPC 模式：使用 `correlation_id` 匹配请求和响应。
90. RPC 模式：为每个客户端创建 exclusive reply queue。
91. RPC 模式：为未完成请求实现超时。
92. Event Sourcing 模式：使用 exchanges 分发事件。
93. Event Sourcing 模式：多个服务可独立消费事件。
94. Event Sourcing 模式：在分区内保持事件顺序。

适用边界 · 注意事项

• 如果系统不适合引入消息中间件，或团队无法运维 RabbitMQ broker，则不要使用本实践。
• 如果需要超大规模日志流、长期回放、分区化流处理，应优先评估 Kafka/Pulsar 等流平台而非 RabbitMQ。
• 如果只需要简单进程内队列或云厂商托管队列且不希望管理 AMQP 细节，则不必按本技能完整实施。