4. 逐笔行情序列号缺口检测与补包触发逻辑

缺口检测的核心对象与判定依据

逐笔行情（逐笔成交、逐笔委托）在交易所侧对每条消息分配单调递增的序列号（不同交易所字段名不同，如上交所逐笔委托的 OrderNo、深交所逐笔的 ApplSeqNum，部分通道还有独立的 ChannelNo 维度）。序列号连续性校验的本质是：对同一通道（或同一标的流）维护一个"期望的下一个序列号"，用实际到达的序列号与之比较。比较结果只有三种：连续、跳变（缺口）、回退（重复或乱序）。

判定时必须先明确两个约定，否则后续逻辑会自相矛盾：

• 期望值语义：expected 表示"下一条应当收到的序列号"，初始化为首条消息的序列号（或快照给出的起始序列号 + 1）。
• 缺口区间语义：收到序列号 seq 时，若 seq > expected，则缺口为左闭右开区间 [expected, seq)，缺口长度 seq - expected。

缺口检测模块本身不负责"补数据"，它只负责发现缺口、上报缺口、触发补包、记录恢复耗时。补包通常走快照服务（如深交所的逐笔快照/重建服务、上交所的行情快照接口），拿到快照后重建订单簿，再把 expected 推进到快照覆盖的序列号之后。

状态机：连续、缺口、补包中、恢复

把检测逻辑写成状态机比写成 if-else 更容易验证边界。四个状态：NORMAL（正常接收）、GAP（已发现缺口、等待补包）、RECOVERING（补包请求已发出、快照未到）、RESYNC（快照到达、重建并推进序列号）。关键约束是：进入 GAP 后，后续到达的增量消息不能直接丢弃，也不能直接应用——正确做法是缓存到待合并队列，等快照重建完成后再按序列号去重合并，否则会丢失缺口之后、快照之前的增量。

检测模块的关键实现

下面给出一个可直接编译运行的 C++17 实现。它把"序列号校验"与"补包触发"解耦：GapDetector 只做判定与状态迁移，补包动作通过回调注入，便于在测试中替换为假实现。时钟使用 std::chrono::steady_clock，避免系统时间跳变污染耗时统计。

#include <cstdint>
#include <chrono>
#include <functional>
#include <string>
#include <vector>
#include <cstdio>

enum class GapState { NORMAL, GAP, RECOVERING, RESYNC };

struct GapEvent {
    uint64_t channel;
    uint64_t expected;      // 缺口左端（含）
    uint64_t received;      // 缺口右端（不含）
    uint64_t gap_len;
    std::chrono::steady_clock::time_point t_gap;
};

class GapDetector {
public:
    using Clock = std::chrono::steady_clock;
    // 补包回调：参数为通道号与缺口区间，返回是否成功受理
    using RequestFn = std::function<bool(uint64_t, uint64_t, uint64_t)>;

    GapDetector(uint64_t channel, uint64_t first_seq, RequestFn req)
        : channel_(channel), expected_(first_seq), req_(std::move(req)) {}

    // 返回 true 表示该消息可立即应用；false 表示已缓存待合并
    bool on_message(uint64_t seq) {
        switch (state_) {
        case GapState::NORMAL:
            if (seq == expected_) { expected_++; return true; }
            if (seq < expected_) { dup_++; return false; }   // 重复/乱序，丢弃
            // seq > expected_：发现缺口
            gap_ = GapEvent{channel_, expected_, seq, seq - expected_, Clock::now()};
            state_ = GapState::GAP;
            pending_.push_back(seq);
            return false;
        case GapState::GAP:
        case GapState::RECOVERING:
            pending_.push_back(seq);   // 缺口后的增量先缓存
            return false;
        case GapState::RESYNC:
            return false;              // 重建期间不应用
        }
        return false;
    }

    // 由外部调度器在 GAP 状态下调用，发出补包请求
    void try_request() {
        if (state_ != GapState::GAP) return;
        if (req_(channel_, gap_.expected, gap_.received)) {
            state_ = GapState::RECOVERING;
            req_count_++;
        } else {
            retry_++;                  // 请求失败，保持 GAP 等待重试
        }
    }

    // 快照到达：snapshot_seq 为快照覆盖到的最大序列号
    void on_snapshot(uint64_t snapshot_seq) {
        if (state_ != GapState::RECOVERING && state_ != GapState::GAP) return;
        state_ = GapState::RESYNC;
        // 快照已覆盖到 snapshot_seq，期望值推进到其后
        expected_ = snapshot_seq + 1;
        // 合并缓存：只保留 seq >= expected_ 的增量，按序去重
        std::vector<uint64_t> merged;
        for (uint64_t s : pending_) {
            if (s >= expected_ && (merged.empty() || s > merged.back()))
                merged.push_back(s);
        }
        pending_.swap(merged);
        if (!pending_.empty()) expected_ = pending_.back() + 1;
        pending_.clear();
        auto t_resync = Clock::now();
        last_recover_us_ = std::chrono::duration_cast<std::chrono::microseconds>(
                               t_resync - gap_.t_gap).count();
        state_ = GapState::NORMAL;
    }

    GapState state() const { return state_; }
    uint64_t expected() const { return expected_; }
    int64_t last_recover_us() const { return last_recover_us_; }
    uint64_t req_count() const { return req_count_; }
    uint64_t retry() const { return retry_; }
    uint64_t dup() const { return dup_; }

private:
    uint64_t channel_;
    uint64_t expected_;
    GapState state_ = GapState::NORMAL;
    GapEvent gap_{};
    std::vector<uint64_t> pending_;
    RequestFn req_;
    int64_t last_recover_us_ = -1;
    uint64_t req_count_ = 0, retry_ = 0, dup_ = 0;
};

使用方式与验证：构造一个通道，首序列号 1001，模拟消息序列 1001, 1002, 1005, 1006, 1007，其中 1003、1004 丢失。预期行为是：1001、1002 立即应用；1005 触发缺口 [1003, 1005)、长度 2；1006、1007 进入缓存；调用 try_request() 后状态变为 RECOVERING；on_snapshot(1004) 后 expected 推进到 1008，缓存中的 1006、1007 被合并，恢复耗时被记录。

#include <cassert>
int main() {
    int requests = 0;
    GapDetector d(1, 1001, [&](uint64_t, uint64_t lo, uint64_t hi) {
        requests++;
        printf("补包请求: 通道1 缺口[%llu, %llu)\n",
               (unsigned long long)lo, (unsigned long long)hi);
        return true;
    });

    assert(d.on_message(1001) == true);
    assert(d.on_message(1002) == true);
    assert(d.on_message(1005) == false);          // 发现缺口
    assert(d.state() == GapState::GAP);
    assert(d.on_message(1006) == false);          // 缓存
    assert(d.on_message(1007) == false);          // 缓存

    d.try_request();
    assert(requests == 1);
    assert(d.state() == GapState::RECOVERING);

    d.on_snapshot(1004);                          // 快照覆盖到 1004
    assert(d.state() == GapState::NORMAL);
    assert(d.expected() == 1008);                 // 1007 之后
    assert(d.last_recover_us() >= 0);
    printf("恢复耗时: %lld us, 补包请求: %llu, 重复丢弃: %llu\n",
           (long long)d.last_recover_us(),
           (unsigned long long)d.req_count(),
           (unsigned long long)d.dup());
    return 0;
}

缺口恢复耗时的观测口径

恢复耗时是本节的核心指标，但它的数值高度依赖口径定义。必须固定以下三点，否则不同批次数据不可比：

建议同时输出三个派生指标：缺口发生率（缺口次数 / 总消息数）、补包成功率（成功重建次数 / 补包请求次数）、恢复耗时 P99。若缺口发生率正常但恢复耗时 P99 偏高，问题通常在快照服务或合并逻辑；若缺口发生率本身偏高，则应先排查网络丢包与解析线程的接收缓冲区。