架构

AI摘要

这是一份关于企业级CDC（变更数据捕获）分发平台的技术架构设计文档。文档详细描述了采用控制面与数据面分离的架构，核心组件包括Laravel-Admin后台、MySQL配置表、Go Admin API、RabbitMQ消息队列、Maxwell Binlog捕获工具及Go Worker消费者。重点阐述了配置驱动的动态路由机制、基于bind_key的队列聚合去重算法、失败重试闭环（含延迟队列与死信队列）以及启动自愈与定时对账的最终一致性保障策略。

配置驱动 · 动态路由 · 企业级 CDC 分发平台架构设计文档

版本：v1.0
更新日期：2026-07-07
状态：设计评审

总体架构概述
控制面（配置驱动层）
- 2.1 后台管理（Laravel-Admin）
- 2.2 MySQL 配置表设计
- 2.3 Go 管理接口（Admin API）
- 2.4 动态聚合与去重算法
- 2.5 RabbitMQ 物理绑定操作
- 2.6 消费者通知机制
- 2.7 启动自愈机制
数据面（消息流转层）
- 3.1 MySQL Binlog 变更捕获（Maxwell）
- 3.2 RabbitMQ Exchange 路由
- 3.3 Queue 消费拉取
- 3.4 Go Worker 消费处理
- 3.5 HTTP 调用下游
- 3.6 失败重试闭环
- 3.7 死信队列与人工兜底
总结与后续方向

1. 总体架构概述

本平台采用 控制面（Control Plane） 与 数据面（Data Plane） 分离的架构思想：

层面	职责	核心组件
控制面	管理订阅规则、动态生成路由、维护 RabbitMQ 绑定关系、保证配置最终一致性	Laravel-Admin、MySQL 配置表、Go Admin API、RabbitMQ 管控
数据面	监听 Binlog、流转消息、执行消费、失败重试、死信兜底	Maxwell、RabbitMQ Exchange/Queue、Go Worker、下游 HTTP 服务

整体数据流向：

MySQL Binlog → Maxwell → RabbitMQ Exchange → (按 Routing Key 路由) → Queue → Go Worker → 下游 HTTP 服务
                                                                                    ↓（失败）
                                                                           延迟重试队列 → 死信队列 → 人工介入

2. 控制面（配置驱动层）

控制面是平台的 “大脑” ，负责将业务人员的订阅意愿，转化为 RabbitMQ 物理路由规则，并保证配置的最终一致性。

2.1 后台管理（Laravel-Admin）

定位：面向运维/业务人员的可视化配置入口。

核心功能：

功能模块	说明
数据源管理	维护 MySQL 实例信息（host、port、binlog 位置），关联 `maxwell_instance` 表
事件定义	定义 `event` 表，记录原始事件元信息（如 `db_name.table_name.operation`）
订阅规则管理	在 `sub_event` 表中创建/修改/删除订阅，指定： - 监听哪个 Event（表 + 操作类型） - 绑定哪个 Consumer（短信/提示/日志等） - 启用/禁用状态
消费者管理	维护 `consumer` 表，记录下游服务信息（HTTP URL、超时时间、重试策略等）
手动触发同步	提供“立即生效”按钮，调用 Go Admin API `/api/refresh_bindings`，将当前配置实时下发至 RabbitMQ

操作示例：

管理员在后台新增一条规则：

字段	值
监听表	`table_user`
操作类型	`insert`
触发动作	发送短信（选择已注册的消费者「短信服务」）
状态	启用

点击保存后，后台写入 sub_event 表，并自动调用 Go API 触发绑定刷新。

2.2 MySQL 配置表设计

说明：以下为核心表结构示意，实际字段可根据业务扩展。

2.2.1 `maxwell_instance` — Maxwell 实例管理

字段名	类型	说明
`id`	INT	主键
`instance_name`	VARCHAR(64)	实例名称（如「订单库-Maxwell」）
`db_host`	VARCHAR(128)	MySQL 主机地址
`db_port`	INT	端口
`db_user`	VARCHAR(32)	同步账号
`db_password`	VARCHAR(128)	密码（加密存储）
`rabbitmq_host`	VARCHAR(128)	对应的 RabbitMQ 主机
`rabbitmq_vhost`	VARCHAR(64)	VHost
`rabbitmq_exchange`	VARCHAR(64)	交换机名称（Maxwell 投递目标）
`status`	TINYINT	1-启用 0-停用

2.2.2 `event` — 事件定义表

字段名	类型	说明
`id`	INT	主键
`db_instance_id`	INT	关联 `maxwell_instance.id`
`routing_key`	VARCHAR(128)	唯一路由键（如 `order_db.table_user.insert`）
`description`	VARCHAR(255)	事件描述（如「用户表新增」）

2.2.3 `consumer` — 消费者定义表

字段名	类型	说明
`id`	INT	主键
`consumer_name`	VARCHAR(64)	消费者名称（如「短信服务」）
`consumer_type`	TINYINT	1-HTTP 2-RPC 3-本地脚本（一期仅支持 HTTP）
`http_method`	VARCHAR(8)	GET/POST
`http_url`	VARCHAR(512)	下游接口地址
`timeout_ms`	INT	超时时间（毫秒）
`retry_times`	TINYINT	最大重试次数（默认 3）
`status`	TINYINT	1-启用 0-停用

2.2.4 `sub_event` — 订阅事件表（核心路由表）

字段名	类型	说明
`id`	INT	主键
`event_id`	INT	关联 `event.id`，决定监听哪个表+操作
`consumer_id`	INT	关联 `consumer.id`，决定触发哪个下游
`bind_key`	VARCHAR(64)	聚合去重键：相同 `bind_key` 的多个订阅共用同一个 Queue（详见 2.4 节）
`status`	TINYINT	1-启用 0-停用
`priority`	TINYINT	优先级（预留）
`created_at`	DATETIME	创建时间
`updated_at`	DATETIME	更新时间

2.2.5 `event_property_setting` — 事件维度属性设置（扩展）

字段名	类型	说明
`id`	INT	主键
`event_id`	INT	关联 `event.id`
`property_name`	VARCHAR(64)	属性名（如 `status`）
`property_operator`	VARCHAR(8)	运算符（`eq`、`ne`、`in` 等）
`property_value`	VARCHAR(255)	属性值（如 `2`）

用途：实现“仅当 status 从 1 变为 2 时才触发”等细粒度过滤。该过滤由 Maxwell 端或 Go Worker 端执行，与 RabbitMQ 路由无关。

2.3 Go 管理接口（Admin API）

定位：控制面的执行引擎，提供 RESTful API 供 Laravel-Admin 及运维工具调用。

2.3.1 核心 API 列表

接口路径	方法	功能
`/api/refresh_bindings`	POST	全量刷新：拉取所有启用订阅，重新计算聚合，更新 RabbitMQ 绑定关系
`/api/refresh_bindings/consumer/{id}`	POST	增量刷新：仅刷新指定消费者相关的绑定
`/api/bindings`	GET	查询当前所有队列及绑定关系（健康检查/对账用）
`/api/reload_config`	POST	通知所有 Go Worker 重新加载本地配置缓存

2.3.2 全量刷新执行流程（`/api/refresh_bindings`）

┌─────────────────────────────────────────────────────────────────────────┐
│ 1. 拉取配置                                                            │
│    SELECT * FROM sub_event                                             │
│    JOIN event ON sub_event.event_id = event.id                        │
│    JOIN consumer ON sub_event.consumer_id = consumer.id               │
│    WHERE sub_event.status = 1 AND consumer.status = 1                 │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 2. 按 (consumer_id + bind_key) 分组聚合                                │
│    例如：                                                              │
│      Group A: consumer=短信服务, bind_key=table_user                  │
│        包含 routing_key: [table_user.insert, table_user.update]       │
│      Group B: consumer=日志服务, bind_key=table_log                   │
│        包含 routing_key: [table_log.delete]                           │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 3. 计算目标状态 (Desired State)                                        │
│    每组生成一个 Queue 定义：                                            │
│      Queue名称: consumer_{consumer_id}.{bind_key}                     │
│      绑定 Keys: 该组所有 routing_key 的集合                           │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 4. 获取 RabbitMQ 当前状态 (Actual State)                               │
│    调用 RabbitMQ Management API 查询：                                 │
│      - 当前存在的队列列表                                               │
│      - 各队列当前的绑定关系                                             │
│    限定前缀：本应用管理的队列（如 queue_prefix = "cdc_"）              │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 5. 计算 Diff (差异)                                                    │
│    - 需要新建的队列：Desired 有，Actual 无                            │
│    - 需要删除的队列：Actual 有，Desired 无（慎重：仅删除明确废弃的）   │
│    - 需要调整绑定的队列：Desired 与 Actual 的 binding 不一致           │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 6. 执行变更（幂等操作）                                                │
│    for each diff:                                                      │
│      6a. 新建队列：Queue.Declare(名称, 死信配置)                      │
│      6b. 增加绑定：Queue.Bind(exchange, routing_key)                  │
│      6c. 删除绑定：Queue.Unbind(exchange, routing_key)               │
│      6d. （可选）删除队列：仅在确认无消息积压且无新订阅时执行          │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 7. 通知消费者（可选）                                                  │
│    调用受影响 Consumer 的 /reload_config 接口，提示重新拉取配置       │
└─────────────────────────────────────────────────────────────────────────┘

2.4 动态聚合与去重算法

问题背景：如果 100 个订阅规则都涉及 table_user（但操作类型不同），不能在 RabbitMQ 中创建 100 个 Queue，否则会浪费资源并导致消息顺序混乱。

解决方案：引入 bind_key 人工/自动聚合。

聚合规则

聚合维度：(consumer_id, bind_key)
同一组内的规则：共享 1 个 Queue，该 Queue 绑定多个 Routing Key
聚合后的队列名称：cdc_{consumer_id}_{bind_key}（加 cdc_ 前缀便于识别和管理）

示例

sub_event	consumer_id	bind_key	routing_key
规则1	1（短信）	`user`	`db.table_user.insert`
规则2	1（短信）	`user`	`db.table_user.update`
规则3	2（日志）	`log`	`db.table_log.insert`
规则4	2（日志）	`log`	`db.table_log.delete`

聚合结果：

Queue 1：cdc_1_user，绑定 db.table_user.insert + db.table_user.update
Queue 2：cdc_2_log，绑定 db.table_log.insert + db.table_log.delete

设计原则：bind_key 建议由后台管理员在创建订阅时人工指定（相同业务域使用相同 bind_key），避免系统自动推断产生误判。

2.5 RabbitMQ 物理绑定操作

Go Admin API 通过 AMQP 协议或 RabbitMQ HTTP API 执行以下操作：

操作	AMQP 方法	说明
声明交换机	`Exchange.Declare`	确保 `maxwell_exchange`（或自定义 Exchange）存在，类型为 `topic`
声明队列	`Queue.Declare`	创建队列，并设置死信参数： `x-dead-letter-exchange` = `retry_exchange` `x-dead-letter-routing-key` = `retry`
绑定路由	`Queue.Bind`	将 Queue 绑定到 Exchange，指定 Routing Key
解绑路由	`Queue.Unbind`	移除 Binding（当订阅被停用时）
删除队列	`Queue.Delete`	仅在队列为空且确认不再需要时执行（通常不主动删除）

关键点：

所有 Queue 必须设置 死信交换机（DLX），这是失败重试闭环的基础（详见 3.6 节）。
绑定操作需具备 幂等性：重复调用不会产生副作用。

2.6 消费者通知机制

当 RabbitMQ 绑定关系发生变化后，需要通知相关消费者服务刷新本地配置，避免消费者拉取到已失效的队列。

通知方式：

方式	说明
主动回调	Go Admin API 在完成绑定后，调用 `consumer.http_url + '/reload'`，通知该消费者重新拉取 `sub_event` 配置
配置中心广播	通过 Redis Pub/Sub 或 Etcd Watch 广播配置变更事件，所有消费者监听并自行 reload
定期轮询（兜底）	消费者自身每隔 5 分钟全量拉取一次配置，防止回调漏掉

一期建议：采用 主动回调 + 定期轮询 双保险。

2.7 启动自愈机制

问题：Go Admin API 重启后，内存中的配置状态丢失；或有人误操作在 RabbitMQ 管理界面删除了绑定，导致物理状态 ≠ 配置状态。

自愈策略：

2.7.1 启动时全量同步（Startup Sync）

Go Admin API 服务启动时，自动执行一次 refresh_bindings 全量流程：

服务启动 → 连接数据库 → 拉取所有启用订阅 → 计算 Desired State → 对比 Actual State → 修正差异

2.7.2 定时对账（Periodic Reconcile）

Go Admin API 启动一个后台定时任务（如每 5 分钟）：

定时器触发 → 全量拉取 DB 配置 → 全量拉取 MQ 绑定 → Diff → 不一致则自动修正 → 记录日志/告警

对账策略：

差异类型	处理动作
DB 有新订阅，MQ 无对应绑定	自动创建 Queue + Bind
DB 已停用订阅，MQ 仍有绑定	自动 Unbind
DB 无该 Queue，MQ 存在（僵尸队列）	记录日志 + 告警（人工确认后删除）

重要：定时对账是 “最终一致性” 的保证，确保任何情况下（网络抖动、人为误操作）配置最终都会收敛到期望状态。

3. 数据面（消息流转层）

数据面是平台的 “肌肉” ，负责从 MySQL 捕获变更，经 RabbitMQ 路由，最终触发下游业务动作。

3.1 MySQL Binlog 变更捕获（Maxwell）

角色：Maxwell 是数据面的起点，伪装成 MySQL 从库，实时接收 Binlog 事件。

3.1.1 Maxwell 部署配置

# config.properties
host=mysql-master.example.com
port=3306
user=maxwell_user
password=******

producer=rabbitmq
rabbitmq_host=rabbitmq.example.com
rabbitmq_vhost=/
rabbitmq_exchange=maxwell_exchange
rabbitmq_exchange_type=topic

# 关键配置：投递时携带 routing_key（格式：{db}.{table}.{type}）
producer_routing_key=%db%.%table%.%type%

# 可选：过滤（通常全量投递，由 RabbitMQ 路由过滤）
filter=exclude: *.*, include: your_db.*

3.1.2 Maxwell 输出的 JSON 格式

INSERT 事件示例：

{
  "database": "order_db",
  "table": "table_user",
  "type": "insert",
  "ts": 1691234567,
  "xid": 12345,
  "commit": true,
  "data": {
    "id": 1001,
    "name": "张三",
    "phone": "13800138000",
    "status": 1
  }
}

UPDATE 事件示例（包含新旧值）：

{
  "database": "order_db",
  "table": "table_user",
  "type": "update",
  "ts": 1691234678,
  "xid": 12346,
  "commit": true,
  "data": {
    "id": 1001,
    "name": "张三",
    "phone": "13800138000",
    "status": 2
  },
  "old": {
    "status": 1
  }
}

DELETE 事件示例：

{
  "database": "order_db",
  "table": "table_user",
  "type": "delete",
  "ts": 1691234789,
  "xid": 12347,
  "commit": true,
  "data": {
    "id": 1001,
    "name": "张三"
  }
}

3.1.3 Routing Key 生成规则

Maxwell 根据配置 producer_routing_key=%db%.%table%.%type% 生成 Routing Key：

事件	Routing Key
`order_db.table_user` 插入	`order_db.table_user.insert`
`order_db.table_user` 更新	`order_db.table_user.update`
`order_db.table_log` 删除	`order_db.table_log.delete`

该 Routing Key 是数据面路由的唯一依据。

3.2 RabbitMQ Exchange 路由

角色：RabbitMQ 的 Topic Exchange 作为 “智能路由器” ，根据 Routing Key 将消息分发到匹配的 Queue。

3.2.1 路由匹配原理

                     ┌─────────────────────────────────────────────┐
                     │         Exchange: maxwell_exchange         │
                     │               (type: topic)                │
                     └─────────────────────────────────────────────┘
                                      │
                ┌─────────────────────┼─────────────────────────────┐
                │                     │                             │
       Routing Key:          Routing Key:                  Routing Key:
       order_db.              order_db.                    order_db.
       table_user.insert      table_user.update            table_log.delete
                │                     │                             │
                ▼                     ▼                             ▼
      ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────┐
      │ Queue: cdc_1_user│  │ Queue: cdc_1_user│  │ Queue: cdc_2_log   │
      │ Binding:         │  │ Binding:         │  │ Binding:           │
      │ *.table_user.*   │  │ *.table_user.*   │  │ *.table_log.*      │
      └─────────────────┘  └─────────────────┘  └─────────────────────┘
               │                     │                      │
               └─────────────────────┴──────────────────────┘
                                     │
                        所有匹配的消息进入各自的 Queue

3.2.2 绑定场景示例

Queue 名称	绑定的 Routing Key	流入的消息
`cdc_1_user`	`.table_user.insert` `.table_user.update`	所有 `table_user` 的 Insert 和 Update
`cdc_1_user`	❌ 未绑定 `*.table_user.delete`	`table_user` 的 Delete 消息被丢弃
`cdc_2_log`	`.table_log.`	`table_log` 的所有操作（Insert/Update/Delete）

核心优势：过滤发生在 RabbitMQ 内部，Queue 里的消息是“纯净”的，Go Worker 无需再做业务判断。

3.3 Queue 消费拉取

角色：Go Worker 作为 RabbitMQ 消费者，从绑定的 Queue 中拉取消息。

3.3.1 消费模式

模式：Pull 模式（Basic.Get）或 Push 模式（Basic.Consume）
推荐：使用 Push 模式 + 手动 ACK，由 RabbitMQ 主动推送给 Worker，并通过 prefetch_count 控制并发度。

3.3.2 并发控制

┌─────────────────────────────────────────────────────────────────────────┐
│  Go Worker 配置                                                       │
│    prefetch_count = 10        // 每次预取 10 条消息，控制并发          │
│    consumer_count = 5         // 启动 5 个协程并发消费                 │
└─────────────────────────────────────────────────────────────────────────┘

3.3.3 消息处理流程（正常路径）

┌─────────────────────────────────────────────────────────────────────────┐
│ 1. 从 Queue 拉取消息（RabbitMQ 推送）                                  │
│    Delivery 包含：Body（JSON）、Routing Key、MessageId 等              │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 2. 解析 JSON，提取关键字段                                             │
│    - database, table, type（操作类型）                                 │
│    - data（变更后的数据）                                              │
│    - old（变更前的数据，仅 update 时有）                              │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 3. 根据 Routing Key 匹配本地订阅规则（缓存）                           │
│    查内存 Map：key = routing_key → consumer_id                        │
│    （本地缓存由 Admin API 的通知或定期轮询维护）                       │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 4. 构造统一 HTTP 请求                                                  │
│    请求体包含：                                                         │
│    {                                                                   │
│      "request_id": "uuid_generated",          // 全局唯一追踪ID        │
│      "channel_id": "cdc_platform",            // 渠道标识               │
│      "event_type": "table_user.insert",       // 事件类型               │
│      "data": { ... },                         // 变更数据               │
│      "old": { ... },                          // 变更前数据（可选）     │
│      "timestamp": 1691234567                  // 事件发生时间           │
│    }                                                                   │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 5. 调用下游 HTTP 接口                                                  │
│    - 方法：POST（统一）                                                │
│    - URL：从 consumer 表获取                                           │
│    - 超时：从 consumer 表获取（如 5000ms）                            │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┴───────────────┐
                    ▼                               ▼
              调用成功                         调用失败/超时
                    │                               │
                    ▼                               ▼
               ACK 消息                    进入失败重试闭环（见 3.6）

3.4 Go Worker 消费处理

3.4.1 本地配置缓存

为避免每次消费都查询数据库，Worker 在内存中维护一份配置缓存：

缓存内容	结构
Routing Key → Consumer	`map[string]Consumer`
Consumer 详细信息	URL、超时、重试次数

缓存更新机制：

更新方式	触发条件
被动刷新	Admin API 调用 `/api/reload_config` 时，Worker 收到通知重新加载
定时轮询	Worker 每隔 5 分钟主动拉取 DB（兜底）
启动加载	Worker 启动时全量拉取一次

3.4.2 请求追踪（Request ID）

所有向下游发起的 HTTP 请求必须携带全局唯一的 request_id（UUID v4），并在日志中记录：

[request_id=abc-123-def] 收到消息 routing_key=order_db.table_user.insert
[request_id=abc-123-def] 调用下游 http://sms.example.com/send 成功 200

用途：全链路追踪、问题定位、日志关联。

3.5 HTTP 调用下游

3.5.1 统一请求规范

字段	类型	必填	说明
`request_id`	string	✅	全局唯一请求 ID（由 Go Worker 生成）
`channel_id`	string	✅	渠道标识（固定值如 `cdc_platform`，用于下游区分来源）
`event_type`	string	✅	完整事件类型（如 `order_db.table_user.insert`）
`data`	object	✅	变更后的数据（JSON 对象）
`old`	object	❌	变更前的数据（仅 UPDATE 时有）
`timestamp`	int64	✅	事件发生时间（Unix 秒级时间戳）

3.5.2 统一返回规范（下游需遵循）

字段	类型	说明
`code`	int	0-成功，非0-失败
`message`	string	错误/成功描述
`data`	object	可选，业务返回数据

Go Worker 处理逻辑：

code == 0：视为成功，ACK 消息
code != 0 或 HTTP 非 2xx 或超时：视为失败，进入重试闭环

3.6 失败重试闭环

核心原则：绝不阻塞主队列。一旦调用下游失败，消息不 Nack（不重新入队），而是被“转移”到延迟重试队列。

3.6.1 架构示意

┌──────────────────────────────────────────────────────────────────────────┐
│                          失败重试闭环架构                               │
│                                                                          │
│  主队列（Main Queue）                                                    │
│       │                                                                  │
│       │ 消费拉取                                                         │
│       ▼                                                                  │
│  ┌─────────┐                                                           │
│  │ Worker  │──调用下游 HTTP                                             │
│  └─────────┘                                                           │
│       │                                                                  │
│       ├── 成功 ──► ACK（结束）                                          │
│       │                                                                  │
│       └── 失败 ──► 发布到延迟重试队列（不 Nack，先 ACK 主队列消息）     │
│                        │                                                 │
│                        ▼                                                 │
│           ┌─────────────────────────────┐                              │
│           │  延迟重试队列（Retry Queue）│                              │
│           │   x-delay: 5s / 30s / 1m    │                              │
│           └─────────────────────────────┘                              │
│                        │                                                 │
│                        │ 延迟到期后自动投递                              │
│                        ▼                                                 │
│           ┌─────────────────────────────┐                              │
│           │     重试消费者               │                              │
│           │  （独立的 Consumer 实例）    │                              │
│           └─────────────────────────────┘                              │
│                        │                                                 │
│                        ├── 成功 ──► ACK                                 │
│                        │                                                 │
│                        └── 失败 & 重试次数 < 3 ──► 再次投递到延迟队列   │
│                        │                                                 │
│                        └── 失败 & 重试次数 >= 3 ──► 投递到死信队列     │
│                                                                          │
│           ┌─────────────────────────────┐                              │
│           │  死信队列（DLQ）             │                              │
│           │  （仅存储，不自动消费）      │                              │
│           └─────────────────────────────┘                              │
│                        │                                                 │
│                        └── 触发告警 + 人工介入                          │
└──────────────────────────────────────────────────────────────────────────┘

3.6.2 延迟重试策略（指数退避）

重试次数	延迟时间	说明
第 1 次	5 秒	快速重试，应对瞬时网络抖动
第 2 次	30 秒	应对下游短暂的 GC 或重启
第 3 次	5 分钟	应对下游较长时间的故障
第 4 次（终局）	—	仍失败则进入死信

延迟时间通过 x-delay 参数设置（需启用 RabbitMQ Delayed Message Plugin）。

3.6.3 重试计数器管理

重试次数存储方式（二选一）：

方案	说明
消息头（Header）	在消息的 `x-retry-count` 头中记录重试次数，每次重试 +1
Redis	以 `{message_id}` 为 key，存储重试次数和原始消息内容

推荐方案：使用消息头，简单可靠，无需额外依赖。

3.6.4 关键注意事项

问题	解决方案
主队列阻塞	主队列消息失败后立即 ACK，不等待重试完成
消息顺序性	同一消息的多次重试走同一延迟队列，自然保持顺序
重试风暴	延迟队列设置了 `x-delay`，不会瞬间冲击下游
重复投递	下游 HTTP 接口必须支持幂等（以 `request_id` 去重）

3.7 死信队列与人工兜底

定位：最后一道防线，确保消息 永不丢失。

3.7.1 死信队列配置

所有主队列在声明时，必须设置以下参数：

x-dead-letter-exchange = retry_exchange
x-dead-letter-routing-key = dlq

当消息被 basic.reject（requeue=false）或 ttl 过期时，自动进入死信队列。

3.7.2 死信处理流程

┌─────────────────────────────────────────────────────────────────────────┐
│ 1. 消息进入死信队列（DLQ）                                             │
│    原因：重试 3 次后仍然失败                                           │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 2. 触发监控告警                                                        │
│    - 钉钉/飞书/企微机器人推送（含失败详情）                            │
│    - 电话告警（P0 级别）                                               │
│    告警内容包含：request_id、routing_key、失败原因、失败时间           │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 3. 人工介入（运维/开发）                                               │
│    - 登录后台查看死信队列消息详情                                      │
│    - 分析失败原因：下游接口问题？数据格式问题？网络问题？              │
│    - 修复问题（如修正下游 Bug、修复脏数据）                           │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 4. 补偿重发                                                            │
│    - 后台提供“死信重投”功能                                            │
│    - 点击重投后，消息重新发布到主队列，进入正常消费流程                │
│    - 或直接调用下游接口进行数据补偿（适用于幂等场景）                  │
└─────────────────────────────────────────────────────────────────────────┘

3.7.3 死信消息存储

死信消息存入数据库（dead_letter_messages 表），以便长期存储和检索：

字段名	类型	说明
`id`	INT	主键
`request_id`	VARCHAR(64)	原始请求 ID
`routing_key`	VARCHAR(128)	路由键
`message_body`	TEXT	完整 JSON 消息
`error_message`	VARCHAR(512)	最后一次失败错误信息
`retry_count`	TINYINT	已重试次数
`status`	TINYINT	0-待处理 1-已修复 2-已忽略
`created_at`	DATETIME	首次失败时间
`updated_at`	DATETIME	最后更新时间

4. 总结与后续方向

4.1 架构优势回顾

特性	实现方式
配置驱动	所有路由规则存于 DB，新增业务只需后台加记录，零代码变更
动态路由	Go Admin API 监听配置变更，动态操作 RabbitMQ Bind/Unbind，实时生效
去重聚合	按 `bind_key` 聚合，避免 RabbitMQ 中大量冗余 Queue
失败隔离	引入延迟重试和死信队列，下游故障不反压 Maxwell 的 Binlog 位点
启动自愈	启动 + 定时双保险同步，保证 MQ 物理状态永远等于 DB 配置状态
全链路追踪	统一 `request_id`，贯穿整个数据流，问题可快速定位

4.2 后续迭代方向（二期/三期）

方向	描述
二次分发	Go Worker 支持将一条消息分发到多个下游（广播模式）
条件过滤下推	将 `event_property_setting` 中的属性条件（如 `status 1→2`）下推到 Maxwell 端，减少无效消息传输
消费者 SDK	提供 Go/Java/PHP SDK，封装统一 HTTP 调用规范，降低下游接入成本
可视化监控	搭建 Grafana 仪表盘，实时展示各 Queue 积压量、消费延迟、重试/死信趋势
消息回放	支持指定时间段内的 Binlog 消息回放，用于故障复盘和数据修复

本文档为架构设计说明，具体实现细节（如接口定义、表结构 DDL）请参考配套的《API 接口文档》和《数据库设计文档》。

本作品采用《CC 协议》，转载必须注明作者和本文链接

米虫

110 声望

记录学习中遇到的问题

0 人点赞

架构