2. 增量消息乱序到达的缓冲重排与缺口检测

乱序缓冲的核心问题

交易所增量消息通常带有单调递增的序列号（如 seq），但网络多路复用、重连补发或推送线程调度会导致消息乱序到达。若直接按到达顺序更新订单簿，会出现“先应用 seq=105 再应用 seq=103”的错误，使价格档位数量、订单引用关系被破坏。本节实现一个乱序缓冲队列：以期望序列号 nextSeq 为基准，把超前到达的消息暂存，按序释放；同时检测缺口并记录，待缺口被填补后自动恢复更新。

缓冲重排的状态机与缺口检测

缓冲器只需维护三个量：nextSeq（下一条应应用的序列号）、buffer（按 seq 索引的暂存表）、gaps（已检测到的缺口区间）。每条消息到达时按以下规则处理：

• seq == nextSeq：直接应用，然后连续从 buffer 中取出 nextSeq+1, nextSeq+2… 一并应用，直到出现空洞。
• seq > nextSeq：说明中间有缺口，记录缺口 [nextSeq, seq-1]，把消息放入 buffer。
• seq < nextSeq：属于重复或过期消息，丢弃并计数（幂等过滤由后续案例专门处理，这里只做丢弃标记）。

缺口填补的判定条件是：当某条消息使 nextSeq 推进到超过缺口右端时，该缺口即被“闭合”，可从 gaps 中移除并记录闭合耗时。

可运行的缓冲重排实现

下面用 Python 实现一个最小但完整的 ReorderBuffer，包含缺口记录、连续释放和闭合统计。示例消息序列与上图一致，便于对照观察。

from dataclasses import dataclass, field
from typing import Dict, List, Tuple

@dataclass
class IncrementalMsg:
    seq: int
    payload: str  # 实际为订单簿增量指令，这里用字符串占位

@dataclass
class ReorderBuffer:
    next_seq: int
    buffer: Dict[int, IncrementalMsg] = field(default_factory=dict)
    gaps: List[Tuple[int, int]] = field(default_factory=list)  # 未闭合缺口 [start, end]
    applied: List[int] = field(default_factory=list)
    dropped: List[int] = field(default_factory=list)
    closed_gaps: List[Tuple[int, int]] = field(default_factory=list)

    def _close_gaps(self):
        """next_seq 推进后，移除右端已被覆盖的缺口"""
        still_open = []
        for start, end in self.gaps:
            if self.next_seq > end:
                self.closed_gaps.append((start, end))
            else:
                still_open.append((start, end))
        self.gaps = still_open

    def _drain(self):
        """从 next_seq 开始连续释放 buffer 中的消息"""
        while self.next_seq in self.buffer:
            msg = self.buffer.pop(self.next_seq)
            self.applied.append(msg.seq)
            self.next_seq += 1
        self._close_gaps()

    def push(self, msg: IncrementalMsg) -> str:
        if msg.seq < self.next_seq:
            self.dropped.append(msg.seq)
            return "dropped"
        if msg.seq == self.next_seq:
            self.applied.append(msg.seq)
            self.next_seq += 1
            self._drain()
            return "applied"
        # msg.seq > next_seq：存在缺口
        if not self.gaps or self.gaps[-1][1] < msg.seq - 1:
            self.gaps.append((self.next_seq, msg.seq - 1))
        self.buffer[msg.seq] = msg
        return "buffered"

# ---- 示例：与图中到达顺序一致 ----
rb = ReorderBuffer(next_seq=101)
arrivals = [101, 103, 105, 102, 104, 106]
for s in arrivals:
    action = rb.push(IncrementalMsg(seq=s, payload=f"delta-{s}"))
    print(f"seq={s:3d} -> {action:8s} next_seq={rb.next_seq} "
          f"gaps={rb.gaps} buffer={sorted(rb.buffer)}")

print("applied:", rb.applied)
print("dropped:", rb.dropped)
print("closed_gaps:", rb.closed_gaps)

预期输出（可直接由上述输入推出）：

seq=101 -> applied  next_seq=102 gaps=[] buffer=[]
seq=103 -> buffered next_seq=102 gaps=[(102, 102)] buffer=[103]
seq=105 -> buffered next_seq=102 gaps=[(102, 102), (104, 104)] buffer=[103, 105]
seq=102 -> applied  next_seq=106 gaps=[] buffer=[]
seq=104 -> dropped  next_seq=106 gaps=[] buffer=[]
seq=106 -> dropped  next_seq=106 gaps=[] buffer=[]
applied: [101, 102, 103, 104, 105, 106]
dropped: []
closed_gaps: [(102, 102), (104, 104)]

缺口检测的边界与验证要点

实现中有三个容易出错的边界，建议在单元测试中逐一覆盖：

验证方法：用上面的示例序列跑一遍，确认 applied 严格等于 [101,102,103,104,105,106]，且 closed_gaps 覆盖所有曾检测到的缺口。再构造“缺口永不填补”的序列（如 101、103、105 后不再有 102），确认 gaps 持续存在且 next_seq 停在 102，此时上层应转入快照重建流程。