事件与消息接口规范

MQ 消息体格式、事件命名约定

事件命名

命名规范

[Requirement] 事件名使用 {领域}.{资源}.{动作} 格式，全小写，点号分隔，动作使用过去式（已发生）或动词原形（命令）。

[Standard] 领域事件（Event）用过去式，表示”已发生”；命令消息（Command）用动词原形，表示”要执行”。

类型	格式	示例
领域事件（Event）	`{domain}.{resource}.{v-past}`	`user.profile.updated`、`order.payment.succeeded`
命令（Command）	`{domain}.{resource}.{verb}`	`order.cancel`、`notification.send`

命名示例

事件名	含义
`user.created`	用户已创建
`user.profile.updated`	用户资料已更新
`order.created`	订单已创建
`order.payment.succeeded`	订单支付成功
`order.payment.failed`	订单支付失败
`inventory.stock.decreased`	库存已扣减
`notification.email.send`	（命令）发送邮件

[Caution] 事件名应表达业务语义，禁止使用技术实现（如 kafka.send、db.insert）。

Topic / Exchange 命名

[Standard] Topic/Exchange 名与事件名一致或为其上级领域，如 order 主题承载 order.* 事件。

[Standard] 消费组命名：{服务名}.{事件名}.consumer，如 order-service.order.payment.succeeded.consumer。

消息体格式

信封结构

[Requirement] 所有 MQ 消息使用统一信封格式，包含元信息与业务数据分离。

{
  "event_id": "550e8400-e29b-41d4-a716-446655440000",
  "event_name": "order.payment.succeeded",
  "event_version": "1.0",
  "event_time": "2026-10-14T08:30:00Z",
  "producer": "payment-service",
  "trace_id": "7f3a9c2e1b4d...",
  "data": {
    "order_id": 1001,
    "amount": 1299,
    "paid_at": "2026-10-14T08:29:58Z"
  }
}

字段	类型	说明
event_id	string	事件唯一 ID（UUID），用于幂等去重
event_name	string	事件名
event_version	string	事件 Schema 版本
event_time	string	事件发生时间（ISO 8601 UTC）
producer	string	生产方服务名
trace_id	string	链路追踪 ID
data	object	业务数据，结构由 event_name + version 决定

业务数据约束

[Standard] data 内字段使用 snake_case，时间用 ISO 8601 UTC，金额用分（整数）。

[Caution] 禁止在消息体中传输大对象（如完整订单明细列表），应只传标识符与必要字段，消费方按需查询。

Schema 管理

[Requirement] 事件 Schema 必须版本化（event_version），不兼容变更需升版本；Schema 注册到 Schema Registry（如 Confluent）。

[Standard] 消息序列化使用 JSON 或 Avro；高吞吐场景使用 Avro + Schema Registry。

幂等性

幂等要求

[Requirement] 消息消费必须保证幂等，即使同一消息被重复投递多次，结果与执行一次一致。

[Standard] 基于 event_id 做去重：消费方处理前检查 event_id 是否已处理，已处理则直接 ACK。

幂等实现

def consume_order_paid(msg):
    event_id = msg["event_id"]
    # 去重锁
    if not redis.set(f"consumed:{event_id}", "1", nx=True, ex=86400):
        logger.info("duplicate event, skip: %s", event_id)
        return  # 已处理，直接确认
 
    try:
        process_order_paid(msg["data"])
    except Exception:
        # 处理失败，删除去重标记，允许重投
        redis.delete(f"consumed:{event_id}")
        raise

[Caution] 业务层也需幂等（如订单状态机、唯一索引），不能仅依赖消息层去重。

事务消息

[Standard] 涉及数据库事务 + MQ 发送的场景，使用事务消息或本地消息表保证最终一致，禁止先提交事务再发消息（可能丢消息）。

本地消息表模式：

业务操作与消息记录写入同一数据库事务。
后台任务扫描未发送消息并投递。
投递成功后标记为已发送。

┌─────────────────────────────────────┐
│ DB 事务                              │
│  1. UPDATE orders SET status='PAID' │
│  2. INSERT INTO outbox_messages...  │  ← 本地消息表
└─────────────────────────────────────┘
         │
         ▼
   后台扫描器 → 投递 MQ → 标记 sent

死信处理

死信队列

[Requirement] 每个消费组必须配置死信队列（DLQ），消息在达到最大重试次数后进入 DLQ，禁止无限重试。

[Default] 最大重试次数 3 次，重试间隔使用指数退避（10s → 1min → 5min）。

死信处理流程

消费失败 → 按退避策略重试。
超过最大重试 → 进入 DLQ。
DLQ 消息保留 7 天，运维可手动重放或丢弃。
DLQ 消息触发告警，人工介入分析。

死信告警

[Requirement] DLQ 消息数量 > 0 即触发告警（企业微信/邮件），告警内容包含：事件名、失败原因、消息摘要、重试次数。

反模式

# 反模式：消费失败直接 ACK，消息丢失
try:
    process(msg)
except Exception:
    logger.error("failed")  # 没有重投，消息直接丢失

# 反模式：无限重试，阻塞队列
while True:
    try:
        process(msg)
        break
    except Exception:
        time.sleep(1)

监控

监控指标

[Requirement] 必须监控以下 MQ 指标并配置告警：

指标	含义	告警阈值
生产消息量（TPS）	每秒生产消息数	突增/突降 > 50%
消费延迟（Lag）	未消费消息堆积数	> 10000 持续 5 分钟
消费失败率	消费失败次数 / 总消费次数	> 1%
DLQ 消息数	死信队列消息数	> 0
消费耗时 P99	单条消息处理耗时	> SLA 的 2 倍

链路追踪

[Standard] 消息必须携带 trace_id，消费方在日志中透传，实现生产→消费全链路追踪。

[Standard] 在 OpenTelemetry 中使用消息中间件自动埋点，串联跨服务调用链。

消费端健康检查

[Default] 提供 /health/mq 端点，检查消费者是否存活、Lag 是否在阈值内，供运维监控与自动扩缩容。

消息可追溯

[Standard] 关键消息（订单、支付）保留原始消息日志（脱敏），保留期 ≥ 30 天，便于排查与对账。