交易所协议分层与序列一致性

1.协议分层 会话层：登录、心跳、重连、序列号维护、重传协商。 业务层：订单簿增量、成交、状态事件、快照事件。 编解码层：字节序转换、字段校验、结构体映射。

#include <cstdint>
#include <optional>
#include
#include <stdexcept>
#include <vector>

enum class SessionMsgType : uint8_t { Logon, Heartbeat, Logout, ReplayRequest };
enum class BizMsgType : uint8_t { Add, Modify, Cancel, Trade, Snapshot };

struct SessionHeader {
  uint64_t seq_num;
  SessionMsgType type;
};

struct BizHeader {
  uint64_t seq_num;
  BizMsgType type;
  uint32_t symbol_id;
};

struct DecodedPacket {
  SessionHeader session;
  BizHeader biz;
  std::span<const std::byte> payload;
};

class Decoder final {
 public:
  [[nodiscard]] static std::optional<DecodedPacket> decode(
      std::span<const std::byte> bytes) noexcept {
    if (bytes.size() < 32) return std::nullopt;
    // 示例：实际项目中应进行协议版本、字段边界、CRC/校验和校验。
    return DecodedPacket{
        .session =
            SessionHeader{.seq_num = 0, .type = SessionMsgType::Heartbeat},
        .biz =
            BizHeader{
                .seq_num = 0, .type = BizMsgType::Snapshot, .symbol_id = 0},
        .payload = bytes.subspan(32)};
  }
};

2.序列一致性处理 每路行情流维护 last_seq，严格检查连续性。 检测到 gap 后进入恢复态：请求重传、合并回放、再切回实时流。 恢复完成前禁止策略继续消费该标的的实时增量。

#include <cstdint>
#include <queue>
#include <stdexcept>

class SeqGate final {
 public:
  enum class State : uint8_t { Live, Recovering };

  explicit SeqGate(uint64_t start_seq = 0)
      : last_seq_(start_seq), state_(State::Live) {}

  [[nodiscard]] bool on_live_packet(uint64_t seq) noexcept {
    if (state_ == State::Recovering) {
      // 恢复态期间实时包只缓存不放行（这里简化为直接拒绝）。
      return false;
    }
    if (seq == last_seq_ + 1) {
      last_seq_ = seq;
      return true;
    }
    if (seq > last_seq_ + 1) {
      state_ = State::Recovering;
      missing_begin_ = last_seq_ + 1;
      missing_end_ = seq - 1;
    }
    return false;
  }

  void on_replay_packet(uint64_t seq) {
    if (state_ != State::Recovering) return;
    if (seq != last_seq_ + 1) {
      throw std::runtime_error("replay packet out-of-order");
    }
    last_seq_ = seq;
    if (last_seq_ == missing_end_) {
      state_ = State::Live;
      missing_begin_ = missing_end_ = 0;
    }
  }

  [[nodiscard]] State state() const noexcept { return state_; }
  [[nodiscard]] uint64_t last_seq() const noexcept { return last_seq_; }
  [[nodiscard]] bool can_trade() const noexcept {
    return state_ == State::Live;
  }

 private:
  uint64_t last_seq_{0};
  uint64_t missing_begin_{0};
  uint64_t missing_end_{0};
  State state_{State::Live};
};

3.时间戳分层 exchange_ts：交易所时间。 nic_rx_ts：网卡接收硬件时间。 app_ts：应用层处理时间。

#include <chrono>
#include <cstdint>

struct TimeTriple {
  uint64_t exchange_ts_ns;  // 交易所报文携带时间戳（ns）
  uint64_t nic_rx_ts_ns;    // NIC 硬件时间戳（ns）
  uint64_t app_ts_ns;       // 应用处理时间戳（ns）
};

[[nodiscard]] inline uint64_t now_ns() noexcept {
  return static_cast<uint64_t>(
      std::chrono::duration_cast<std::chrono::nanoseconds>(
          std::chrono::steady_clock::now().time_since_epoch())
          .count());
}

[[nodiscard]] inline int64_t wire_to_app_latency_ns(
    const TimeTriple& t) noexcept {
  return static_cast<int64_t>(t.app_ts_ns) -
         static_cast<int64_t>(t.nic_rx_ts_ns);
}

[[nodiscard]] inline int64_t exch_to_nic_latency_ns(
    const TimeTriple& t) noexcept {
  return static_cast<int64_t>(t.nic_rx_ts_ns) -
         static_cast<int64_t>(t.exchange_ts_ns);
}