陈望舒
已采纳
结论：把外部副作用纳入幂等状态机，用"副作用日志表 + 乐观锁"替代"幂等表 + 频控"。
你当前的核心矛盾是：幂等表管的是"业务写入"，WMS/SMS 是"外部副作用"，两者幂等粒度不一致且副作用不可回滚。解法不是加频控，而是把副作用本身纳入一个可查询、可重试、可审计的状态机。
一、副作用日志表设计
CREATE TABLE side_effect_log (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
biz_id VARCHAR(64) NOT NULL COMMENT '业务幂等键，如 order_id',
effect_type ENUM('WMS_PICK','SMS_NOTIFY') NOT NULL,
idempotency_key VARCHAR(128) NOT NULL COMMENT '下游幂等键，如 WMS request_id',
status TINYINT NOT NULL DEFAULT 0
COMMENT '0=PENDING 1=IN_PROGRESS 2=DONE 3=FAILED',
attempt INT NOT NULL DEFAULT 0,
last_error VARCHAR(512),
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
UNIQUE KEY uk_biz_effect (biz_id, effect_type),
INDEX idx_status (status, updated_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
关键：uk_biz_effect 唯一键是"副作用幂等锚点"——同一订单的同一类副作用只允许一条记录，INSERT 冲突即代表"已经有人处理过"。
二、消费逻辑改造（伪代码）
def consume(msg):
biz_id = msg.order_id
# Step 1-2: 业务写入（现有幂等表逻辑不变）
if not idempotent_check_and_write(biz_id):
return
# Step 3-4: 外部副作用走状态机
ensure_side_effect(biz_id, 'WMS_PICK', f"wms-{biz_id}")
ensure_side_effect(biz_id, 'SMS_NOTIFY', f"sms-{biz_id}")
commit_offset()
def ensure_side_effect(biz_id, effect_type, request_id):
# 1) INSERT（利用唯一键做天然幂等）
try:
db.execute("INSERT INTO side_effect_log(biz_id,effect_type,idempotency_key) VALUES(?,?,?)",
(biz_id, effect_type, request_id))
except DuplicateKeyError:
pass # 已存在
# 2) CAS 抢占：PENDING(0) → IN_PROGRESS(1)
affected = db.execute(
"UPDATE side_effect_log SET status=1, attempt=attempt+1 WHERE biz_id=? AND effect_type=? AND status=0",
(biz_id, effect_type))
if affected == 0:
row = db.query_one("SELECT status, updated_at FROM side_effect_log WHERE biz_id=? AND effect_type=?",
(biz_id, effect_type))
if row.status == 2:
return # 已完成，直接跳过
if row.updated_at < now() - timedelta(seconds=30):
# 超时残留（上次崩溃），回退后重新抢占
db.execute("UPDATE side_effect_log SET status=0 WHERE biz_id=? AND effect_type=? AND status=1 AND updated_at < ?",
(biz_id, effect_type, now() - timedelta(seconds=30)))
affected = db.execute("UPDATE side_effect_log SET status=1, attempt=attempt+1 WHERE biz_id=? AND effect_type=? AND status=0",
(biz_id, effect_type))
else:
return # 正在处理中，本轮跳过
# 3) 执行外部调用
try:
if effect_type == 'WMS_PICK':
call_wms_api(request_id)
else:
call_sms_api(request_id)
# 4) 标记完成
db.execute("UPDATE side_effect_log SET status=2 WHERE biz_id=? AND effect_type=?", (biz_id, effect_type))
except Exception as e:
db.execute("UPDATE side_effect_log SET status=3, last_error=? WHERE biz_id=? AND effect_type=?",
(str(e)[:512], biz_id, effect_type))
raise # 让 Kafka 重试或进 DLQ
三、为什么比"频控 + outbox"更稳
| 维度 | 频控 | outbox 轮询 | 副作用日志表+状态机 | |---|---|---|---| | 幂等粒度 | 时间窗口（粗糙） | 业务键 | 业务键 | | 并发安全 | 依赖外部锁 | 依赖轮询间隔 | CAS + 唯一键 | | 可审计 | 无 | 有 | 有（含 attempt/error） | | 失败重试 | 无 | 有 | 有（FAILED + 定时扫描） | | 额外延迟 | 无 | 2s 轮询 | 无（同步调用） | | 新增故障点 | 少 | 多（worker 进程） | 少（纯 DB 状态） |
四、SMS 等不可逆副作用的兜底
SMS 一旦发出就收不回，额外加两层：
side_effect_log里记录effect_payload（模板 + 变量），重发走人工审批而非自动重发。- SMS 网关的频控从"静默拦截"改为"告警 + 放行"：命中频控时写
sms_suppress记录并通知运营，由运营决定是否补发。这样"用户投诉后人工重发"不再被误拦。
五、定时兜底（防 IN_PROGRESS 卡死）
-- 每 30s 跑一次，回退超时 60s 且重试未超限的记录
UPDATE side_effect_log
SET status = 0, updated_at = NOW(3)
WHERE status = 1
AND updated_at < NOW(3) - INTERVAL 60 SECOND
AND attempt < 3;
超过 3 次的进死信队列人工介入，避免无限重试打爆下游。
六、一句话
幂等表管"写不写"，副作用日志表管"调不调"，状态机管"调没调完"。三者各司其职，rebalance 后新实例查状态就知道该不该再调——不存在"幂等表放行了但副作用已经发了"的窗口。
赞 0