订单簿重建引擎的架构演进与性能优化

• 步骤一：缓冲区初始化 (Prepare)
  1. 监控系统识别出热点股票 S 需从 Batch A 迁移至 Batch B。
  2. 系统向 Batch B 发送 MigrateStartCmd。
  3. Batch B 收到指令后，为股票 S 初始化暂存缓冲区（Buffer）。此时，Batch B 并不处理股票 S 的逻辑，仅接收并缓存后续到达的数据。

• 步骤二：路由原子切换 (Switch)
  1. 系统更新全局路由表。
  2. 此后到达的股票 S 的所有新数据（Seq  N+1 以后）将被路由至 Batch B 并进入其缓冲区。

• 步骤三：源线程极速排空 (Fast Drain)
  1. 系统向 Batch A 发送 MigrateFlushCmd，该指令排在 Batch A 当前积压的所有旧数据（Seq N 以前）之后。
  2. 优化关键点：系统同时通知 Batch A 进入“打包模式”。Batch A 不再对股票 S 的残留数据执行高耗时的订单簿计算，而是直接将其从队列中取出并存储到LegacyData 列表中。
  3. 此举将 Batch A 的处理耗时从  O(计算) 降级为  O(移动)，极大缩短了积压数据的排空时间，快速释放 Batch A 的算力。
• 步骤四：状态接力与回放 (Handover & Replay)
  1. 当 Batch A 处理到 MigrateFlushCmd 时，意味着旧数据已全部打包。Batch A 发送包含 LegacyData 列表数据的 MigrateCompleteCmd 至 Batch B。
  2. Batch B 收到指令后，执行严格顺序的回放操作：
     1. Replay Legacy：处理 Batch A 移交的旧数据，更新订单簿状态至 Seq N。
     2. Replay Buffer：处理缓冲区内积压的新数据（Seq N+1 ...）。
     3. Resume：移除缓冲区，恢复正常的实时计算模式。

• 零锁竞争：全流程无 lock 或 WaitHandle 阻塞，完全依赖队列消息传递，最大化了 CPU 利用率。
• 顺序保证：利用 TCP/IP 类似的滑动窗口思想，通过消息传递确保了 Seq N 与 Seq N+1 的无缝衔接。
• 性能隔离：通过“打包模式”快速释放源线程压力，避免了迁移过程对源线程中其他股票的持续影响。

4.具体实现

核心思想就是通过引入三种控制指令（Command），利用消息队列的 FIFO 特性来协调迁移，全程无锁。

1. 定义控制指令：为了在数据流中插入控制逻辑，我们定义了三种特殊消息：

   // 指令1：通知目标线程（Target）建立缓冲区
   class MigrateStartCmd { public string Symbol; }
   
   // 指令2：通知源线程（Source）停止计算，开始打包排空
   class MigrateFlushCmd { public string Symbol; public int TargetBatchId; }
   
   // 指令3：源线程发给目标线程的交接数据包
   class MigrateCompleteCmd 
   {
       public string Symbol;
       public List<object> LegacyData; // 核心优化：打包好的旧数据
   }

2. 调度器逻辑 (FastMarketDataManager)，调度器负责编排迁移的四个原子步骤：准备 -> 降级 -> 切换 -> 哨兵。

   /// <summary>
   /// 新建批次并移动股票
   /// </summary>
   public bool MoveStockToNewBatch(string symbol)
   {
       lock (postProcessors)
       {
           int newBatchId = postProcessors.Count;
           //这里传入 newBatchId 作为 ID
           FastMDPostProcessor newProcessor = new FastMDPostProcessor(newBatchId, fastMdDic, postProcessors, config.LogFastMDError);
           postProcessors.Add(newProcessor);
   
           if (MoveStockToBatch(symbol, newBatchId))
           {
               return true;
           }
           return false;
       }
   }
   
   public void MoveStockToBatch(string symbol, int targetBatchId)
   {
       int currentBatchId = symbolBatchIndex[symbol];
   
       // Step 1: 通知 Target 开启缓冲模式
       // Target 收到此指令后，会暂存该股票的新数据，不进行计算
       postProcessors[targetBatchId].Enqueue(new MigrateStartCmd(symbol));
   
       // Step 2: 通知 Source 开启“打包模式” (Direct Flag)
       // 这一步至关重要：Source 线程后续不再执行高耗时的 OrderBook 计算，
       // 而是直接将数据 Add 到 List，极大提升排空速度。
       postProcessors[currentBatchId].MarkStockAsMigratingOut(symbol);
   
       // Step 3: 原子切换路由
       // 此后的新数据流向 Target
       symbolBatchIndex[symbol] = targetBatchId;
   
       // Step 4: 发送哨兵指令
       // Source 处理完当前队列积压的数据后，会遇到此指令，触发移交
       postProcessors[currentBatchId].Enqueue(new MigrateFlushCmd(symbol, targetBatchId));
   }

3. 处理器逻辑 (FastMDPostProcessor)，消费者线程的循环逻辑升级为支持状态机的处理模式。

   private void PostProcess()
   {
       while (postProcessQueue.TryDequeue(out object obj))
       {
           string symbol = GetSymbol(obj);
   
           // === 场景 A: 正在移入 (Target) ===
           // 新数据到达 Target，但旧数据还没来，先缓存
           if (incomingBuffers.TryGetValue(symbol, out var buffer))
           {
               buffer.Enqueue(obj);
               continue;
           }
   
           // === 场景 B: 正在移出 (Source) ===
           // 核心优化：降级处理。不计算盘口，直接打包，加速队列排空
           if (outgoingBuffers.TryGetValue(symbol, out var legacyList))
           {
               legacyList.Add(obj);
               continue;
           }
   
           // === 场景 C: 正常计算 ===
           if (obj is TAS tas) ProcessTAS(tas);
   
           // === 场景 D: 指令处理 ===
           else if (obj is MigrateFlushCmd cmd)
           {
               // Source 完成打包，发送数据包给 Target
               var legacyData = outgoingBuffers[cmd.Symbol];
               var target = allProcessors[cmd.TargetBatchId];
               target.Enqueue(new MigrateCompleteCmd(cmd.Symbol, legacyData));
           }
           else if (obj is MigrateCompleteCmd cmd)
           {
               // Target 收到旧数据包，开始严格顺序回放
               // 1. 回放 Source 的旧数据
               foreach(var item in cmd.LegacyData) Process(item);
               // 2. 回放 Buffer 中的新数据
               while(incomingBuffers[cmd.Symbol].Count > 0) 
                   Process(incomingBuffers[cmd.Symbol].Dequeue());
               // 3. 恢复正常
               incomingBuffers.Remove(cmd.Symbol);
           }
       }
   }

四、验证与测试

为了验证这套复杂状态机的正确性， 构建了名为 MockDataFeed的数据类。
Mock 策略：

1. 确定性数据源：生成带有严格递增序列号 (SN) 的 Mock TAS 数据。
2. 热点模拟：通过 SetHotSpot(symbol, true) 方法，瞬间将某只股票的数据生成速率提升 2000 倍，人为制造队列积压。
3. 一致性断言：在迁移发生时，通过日志系统追踪 SN 的连续性。
   • Expect: Batch A 处理 SN: 1000 -> Batch B 处理 SN: 1001。
   • Actual: 验证通过。

代码如下：

class MockDataFeed : DataFeedAdaptor, IFastMarketDataPendingReporter
{
    /// <summary>
    /// 内部类：维护单个模拟股票的状态
    /// </summary>
    class MockStockContext
    {
        public string Symbol { get; }
        public ExchangeType Exchange { get; }
        public double LastPrice { get; set; } = 10.0;
        public long CurrentSN { get; set; } = 0; // 维护严格递增的序列号
        public volatile bool IsHotSpot = false; // 是否开启洪水模式

        public MockStockContext(string symbol, ExchangeType exchange)
        {
            Symbol = symbol;
            Exchange = exchange;
        }
    }

    private string name;

    private long receivedData;
    private FastMarketDataManager fastMarketDataManager;
    private readonly XTPConfig config;

    private volatile bool _running;
    private Thread _generatorThread;
    private readonly List<MockStockContext> _stocks;
    private readonly Random _random = new Random();

    // 配置参数
    private const int StockCount = 40; // 模拟20只股票
    private const int NormalSpeed = 1; // 普通股票每次循环生成1条数据
    private const int HotSpotSpeed = 2000; // 热点股票每次循环生成2000条数据 (制造积压)
                                           // 新增：每生成多少笔TAS后，生成一笔MarketData快照
    private const int MarketDataInterval = 10;
    public MockDataFeed(string configName)
    {
        name = configName;
        config = new XTPConfig(configName);

        FastMarketDataConfig cfg = new FastMarketDataConfig(
            config.FastMDUpdateInterval,
            config.CallAuctionFastMDUpdateInterval,
            config.SZFastMDUpdateDelay,
            config.SHFastMDUpdateDelay,
            config.ProcessFastMarketDataBatchCount,
            config.LogFastMDError);
        fastMarketDataManager = new FastMarketDataManager(cfg);
        fastMarketDataManager.NewFastMarketDatas += (mds) =>
        {
            // 可以在这里打断点或日志观察生成的 MarketData
            // Logger.LogInfo($"Output {mds.Count} MarketDatas");
        };

        // 3. 初始化模拟股票列表
        _stocks = new List<MockStockContext>();
        for (int i = 0; i < StockCount; i++)
        {
            // 混合生成沪深股票
            string prefix = i % 2 == 0 ? "600" : "000";
            string symbol = $"{prefix}{i.ToString("D3")}";
            // 交易所类型
            var exchange = i % 2 == 0 ? ExchangeType.SH : ExchangeType.SZ;

            _stocks.Add(new MockStockContext(symbol, exchange));
        }
        SymbolFilter.Instance.SetFastSymbolForDebug(_stocks.Select(x => x.Symbol).ToList());
    }

    public override string Name
    {
        get { return name; }
    }

    public override long ReceivedData
    {
        get { return receivedData; }
    }

    public override void Start()
    {
        if (_running) return;
        _running = true;

        // 启动生成线程
        _generatorThread = new Thread(DataGeneratorWork)
        {
            Name = "MockDataGenerator",
            IsBackground = true
        };
        _generatorThread.Start();
        Logger.LogInfo("MockDataFeed Started. Generating data...");
    }

    public override void Stop()
    {
        _running = false;
        if (_generatorThread != null && _generatorThread.IsAlive)
        {
            _generatorThread.Join(2000);
        }
        fastMarketDataManager.Stop();
    }

    /// <summary>
    /// 数据生成主循环
    /// </summary>
    private void DataGeneratorWork()
    {
        // 先推送一波初始的 MarketData，确保 FastMarketData 对象被创建
        foreach (var stock in _stocks)
        {
            var md = GenerateMarketData(stock);
            fastMarketDataManager.UpdateMD(md);
        }
        while (_running)
        {
            try
            {
                foreach (var stock in _stocks)
                {
                    // 决定本次生成的数据量
                    int count = stock.IsHotSpot ? HotSpotSpeed : NormalSpeed;

                    for (int i = 0; i < count; i++)
                    {
                        // 随机生成 TAS 或 OrderData
                        // 为了测试顺序性，我们主要生成 TAS，因为 TAS 有明确的 Sequence Number (SN)
                        TAS tas = GenerateNextTAS(stock);
                        // 推送给 Manager
                        fastMarketDataManager.UpdateTAS(tas);
                    }
                }

                // 稍微休眠一下，控制整体节奏，避免 CPU 100%
                // 如果有热点股票，这个 Sleep 会导致积压（因为生成速度远大于消费速度）
                Thread.Sleep(100);
            }
            catch (Exception ex)
            {
                Logger.LogError($"Mock Generator Error: {ex.Message}");
            }
        }
    }

    private MarketData GenerateMarketData(MockStockContext stock)
    {
        double currentPrice = Math.Round(stock.LastPrice, 2);

        MarketData md = new MarketData
        {
            Symbol = stock.Symbol,
            Exchange = stock.Exchange,
            Time = DateTime.Now,
            CurrentPrice = currentPrice,
            Type = InstrumentType.Stock, // 确保类型正确
        };

        return md;
    }
    private TAS GenerateNextTAS(MockStockContext stock)
    {
        // 价格随机波动
        double change = (_random.NextDouble() - 0.5) * 0.1;
        stock.LastPrice = Math.Max(0.01, stock.LastPrice + change);
        // 严格递增 SN
        stock.CurrentSN++;
        TAS tas = new TAS(stock.Symbol)
        {
            Exchange = stock.Exchange,
            Time = DateTime.Now,
            Price = Math.Round(stock.LastPrice, 2),
            Size = _random.Next(100, 10000),
            SN = stock.CurrentSN, // 关键：用于验证顺序
            TradeType = TASTradeType.Trade,
            BSFlag = TAS.BSType.Buy // 简单模拟
        };

        return tas;
    }

    public override string GetStatus()
    {
        return base.GetStatus();
    }

    /// <summary>
    /// 移动股票到指定批次 (对接 UI)
    /// </summary>
    public void MoveStockToBatch(string symbol, int targetBatchID)
    {
        Logger.LogInfo($"UI Command: Move {symbol} to Batch {targetBatchID}");
        fastMarketDataManager.MoveStockToBatch(symbol, targetBatchID);
    }

    /// <summary>
    /// 设置某只股票是否为热点 (用于制造积压)
    /// </summary>
    public void SetStockHotSpot(string symbol, bool isHot)
    {
        var stock = _stocks.FirstOrDefault(s => s.Symbol == symbol);
        if (stock != null)
        {
            stock.IsHotSpot = isHot;
            Logger.LogInfo($"Stock {symbol} HotSpot set to: {isHot}");
        }
    }

    public List<BatchPendingInfo> GetPendingInfo()
    {
        if (fastMarketDataManager != null)
        {
            return fastMarketDataManager.GetPendingInfo();
        }
        return new List<BatchPendingInfo>();
    }

}

这里面，提供了将某只股票设置为热点和取消热点的方法，提供了将股票从一个热点移动到另外一个热点的方法。

五：可观测的结果

最后，设计了一个界面，根据我们预想的，可以看到新的结构能够处理好方案二中的静态哈希可能产生的负载倾斜，达到了预期的效果。

▲ 正常情况下，处理批次，以及积压情况

▲将批次0中的600030股票设置为热点股票，可以看到批次0的积压量迅速上升，而且还会影响到同批次里面的其它股票产生积压，但是目前不会对其它批次产生影响

▲ 将批次0中的热点股票600030，移动到空闲的批次4中，可以看到，批次0很快恢复正常，但会导致批次4产生新的积压。

▲再次将600030从批次4移动到新的批次（方法里传参数-1），可以看到新建了批次5，里面只有一只股票。批次5产生了积压，其它批次恢复正常。

参考：

• https://mp.weixin.qq.com/s/WsyCqs51TocXvEprGQnbmQ 
• https://findata-insight.htsc.com:9151/insight_help/exquisiteArticle/OrderSheet/#_1 
• https://docs.dolphindb.cn/zh/tutorials/orderBookSnapshotEngine.html