4. 序列号跳变时的订单簿重建触发策略

跳变检测的核心判断维度

序列号跳变与普通乱序不同：乱序是消息迟到但最终会到，跳变意味着中间一段序列号可能永久缺失。处理策略不能一刀切，需要同时看两个量：跳变幅度（缺口大小）和持续时间（缺口存在多久未被填补）。幅度小且短暂，通常只是网络抖动，等待补发即可；幅度大或持续超时，则说明数据源可能丢包或推送中断，必须请求新快照重建。

检测逻辑实现

核心思路：维护一个expectedSeq（下一条应到达的序列号）。每收到一条增量消息，比较其序列号与expectedSeq。若相等则正常应用并递增；若大于则记录缺口，启动计时器；若小于则属于重复消息（由幂等过滤器处理，本例不展开）。缺口在计时器到期前被填补则清除，否则触发快照重建。

import time
from dataclasses import dataclass, field
from enum import Enum

class Decision(Enum):
    APPLY = "apply"
    WAIT = "wait"
    REBUILD = "rebuild"

@dataclass
class GapContext:
    expected_seq: int
    received_seq: int
    gap_size: int
    first_seen_ms: float
    decision: str = ""
    reason: str = ""

class SequenceGapDetector:
    def __init__(self, max_gap_size=50, max_gap_age_ms=2000):
        self.expected_seq = None
        self.max_gap_size = max_gap_size
        self.max_gap_age_ms = max_gap_age_ms
        self.active_gap: GapContext | None = None
        self.decision_log: list[GapContext] = []

    def on_message(self, seq: int, now_ms: float | None = None) -> Decision:
        now_ms = now_ms if now_ms is not None else time.monotonic() * 1000

        if self.expected_seq is None:
            self.expected_seq = seq + 1
            return Decision.APPLY

        if seq == self.expected_seq:
            # 缺口被填补
            if self.active_gap:
                self.active_gap.decision = "filled"
                self.decision_log.append(self.active_gap)
                self.active_gap = None
            self.expected_seq = seq + 1
            return Decision.APPLY

        if seq < self.expected_seq:
            return Decision.APPLY  # 重复消息，交给幂等层

        # seq > expected_seq：出现跳变
        gap_size = seq - self.expected_seq
        if self.active_gap is None:
            self.active_gap = GapContext(
                expected_seq=self.expected_seq,
                received_seq=seq,
                gap_size=gap_size,
                first_seen_ms=now_ms,
            )

        gap_age = now_ms - self.active_gap.first_seen_ms

        # 决策：幅度大 或 持续超时 → 重建
        if gap_size > self.max_gap_size or gap_age > self.max_gap_age_ms:
            self.active_gap.decision = "rebuild"
            self.active_gap.reason = (
                f"gap_size={gap_size}>{self.max_gap_size}"
                if gap_size > self.max_gap_size
                else f"gap_age={gap_age:.0f}ms>{self.max_gap_age_ms}ms"
            )
            self.decision_log.append(self.active_gap)
            self.active_gap = None
            return Decision.REBUILD

        return Decision.WAIT

    def tick(self, now_ms: float) -> Decision:
        """定时调用，检查活跃缺口是否超时"""
        if self.active_gap is None:
            return Decision.APPLY
        gap_age = now_ms - self.active_gap.first_seen_ms
        if gap_age > self.max_gap_age_ms:
            self.active_gap.decision = "rebuild"
            self.active_gap.reason = f"gap_age={gap_age:.0f}ms>{self.max_gap_age_ms}ms"
            self.decision_log.append(self.active_gap)
            self.active_gap = None
            return Decision.REBUILD
        return Decision.WAIT

验证与判读

用一组构造的序列号流验证三种路径：小缺口被填补、大缺口立即重建、小缺口超时升级重建。以下为教学示例输入与预期决策，非实测结果。