Storm 交易系統高可靠消息處理：避免重複消費的 Redis 去重 + 異常/超時雙重防護架構

業務背景與架構演進

實時交易系統，平時 QPS 1000，雙十一峰值數十萬。採用 Kafka + Storm 架構：交易數據寫入 Kafka → Topology A 消費處理規則匹配 → 推送不同通知主題 → Topology B 消費推送至客戶端（微信/支付寶）。

核心配置參數

• Storm 超時時間：topology.message.timeout.secs = 180 (3分鐘)
• KafkaSpout pending 队列：spout.pending.max = 2000
• 啟用 ACK 機制：異常時主動 ack，防止無限重試
• 每筆交易擁有全局唯一 ID (UUID)

初版去重設計：Topology A 單層 Redis 去重

OnceBolt 邏輯

public void execute(Tuple tuple) {
    String transactionId = tuple.getStringByField("transactionId");
    // Redis SETNX 原子去重
    if (redis.setnx("tx:processed:" + transactionId, "1")) {
        redis.expire("tx:processed:" + transactionId, 86400); // 24h 過期
        collector.emit(tuple); // 轉發給 CalculateBolt
    } else {
        // 已處理，直接 ack 不再下發
        collector.ack(tuple);
    }
}

CalculateBolt 匹配規則 → 推送通知 Kafka 主題。

線上事故：Topology B 仍出現重複推送

現象

運行半年後，用戶收到重複通知（2-3 條相同內容）。

根因分析：Storm Tuple 超時導致的重複處理

1. Topology A 正常處理，OnceBolt 去重生效，CalculateBolt 處理完成，推送 Kafka
2. Topology B 消費通知主題，推送至第三方接口（微信/支付寶）
3. 下游接口響應慢/網絡抖動，導致單 Tuple 處理時間 > 3 分鐘
4. Storm 判定 Tuple 超時失敗，Spout 重新發射該 Tuple
5. 原 Tuple 並未死亡，資源釋放後繼續處理完成 → 雙重處理
6. Topology B 缺乏去重機制，導致重複推送

關鍵認知

Storm ACK 機制只能保證 至少一次 語義。超時重發 + 原請求未真正失敗 = 重複處理。業務層冪等性/去重是唯一根治方案。

終極解決方案：全鏈路去重 + 異常/超時區分處理

方案一：Topology B 補充去重（快速止血）

在 Topology B 入口 Bolt 添加相同 Redis 去重邏輯，Key 設計：notify:processed:{notificationId}。

方案二：業務級冪等設計（徹底治本）

通知消息攜帶 bizId（如訂單號+通知類型），下游推送接口基於 bizId 冪等處理（數據庫唯一索引/Redis SETNX）。

方案三：智能區分異常重試 vs 超時重發（進階優化）

public class SmartAckBolt extends BaseRichBolt {
    @Override
    public void execute(Tuple tuple) {
        String txId = tuple.getStringByField("transactionId");
        try {
            // 業務處理
            processBusiness(tuple);
            // 成功：標記為正常完成，防止超時重發時再次處理
            redis.setnx("tx:success:" + txId, "1");
            redis.expire("tx:success:" + txId, 86400);
            collector.ack(tuple);
        } catch (BusinessException e) {
            // 業務異常：標記為異常，允許重試
            redis.setnx("tx:failed:" + txId, "1");
            collector.ack(tuple); // 不重試業務邏輯錯誤
        } catch (Exception e) {
            // 系統異常：不 ack，讓 Storm 重試
            collector.fail(tuple);
        }
    }
}

// OnceBolt 升級版：三態判斷
public void execute(Tuple tuple) {
    String txId = tuple.getStringByField("transactionId");
    if (redis.exists("tx:success:" + txId)) {
        // 正常完成態：超時重發導致的重複，直接丟棄
        collector.ack(tuple);
        return;
    }
    if (redis.exists("tx:failed:" + txId)) {
        // 異常失敗態：允許重試處理
        collector.emit(tuple);
        return;
    }
    // 首次處理態：執行去重鎖
    if (redis.setnx("tx:processing:" + txId, "1")) {
        redis.expire("tx:processing:" + txId, 300); // 5分鐘處理窗口
        collector.emit(tuple);
    } else {
        collector.ack(tuple); // 並發處理中，丟棄重複
    }
}

架構對比總結

生產環境優化建議

1. 調大超時時間：根據 P99 延遲設置 topology.message.timeout.secs，預留 2-3 倍餘量
2. 監控 Tuple 處理時間：接入 Storm UI / Metrics 監控，告警慢 Tuple
3. 下游接口熔斷：Hystrix/Sentinel 保護第三方調用，快速失敗觸發 Storm 重試而非拖慢
4. Redis 連接池調優：去重 Bolt 併發高，配置足夠連接數、超時、重試
5. Key 過期策略：根據業務留存周期設置 TTL，防止 Redis 內存洩漏
6. 冪等 Key 設計：bizType:bizId 確保全局唯一，避免不同業務衝突

總結

Storm 高可靠 ≠ 無重複。ACK 機制解決「不丟失」，冪等去重解決「不重複」。交易系統通過 Redis 多態去重 + 下游冪等接口 雙重保障，實現「消息必達、業務不重」的強一致性體驗。架構設計時必須考慮 超時重發導致的雙重執行 這一分布式系統固有特性，而非僅關注異常重試。