本章主要通过源码阅读，解读一下CppTrader项目的内部架构设计，主要包括以下几个模块:

1. NASDAQ ITCH协议处理器；

NASDAQ ITCH协议处理器

在了解ITCH协议处理器之前，首先需要了解以下ITCH协议前置知识：ITCH协议参考

ITCH处理器头文件设计

#ifndef CPPTRADER_ITCH_HANDLER_H
#define CPPTRADER_ITCH_HANDLER_H

#include "utility/endian.h"
#include "utility/iostream.h"

#include <vector>

namespace CppTrader {
/*!
    \namespace CppTrader::ITCH
    \brief ITCH protocol definitions
*/
namespace ITCH {

//! System Event Message 系统事件消息
struct SystemEventMessage{};
//! Stock Directory Message
struct StockDirectoryMessage{};
//! Stock Trading Action Message
struct StockTradingActionMessage{};
//! Reg SHO Short Sale Price Test Restricted Indicator Message
struct RegSHOMessage{};
//! Market Participant Position Message
struct MarketParticipantPositionMessage{};
//! MWCB Decline Level Message
struct MWCBDeclineMessage{};
//! MWCB Status Message
struct MWCBStatusMessage{};
//! IPO Quoting Period Update Message
struct IPOQuotingMessage{};
//! Add Order Message
struct AddOrderMessage{};
//! Add Order with MPID Attribution Message
struct AddOrderMPIDMessage{};
//! Order Executed Message
struct OrderExecutedMessage{};
//! Order Executed With Price Message
struct OrderExecutedWithPriceMessage{};
//! Order Cancel Message
struct OrderCancelMessage{};
//! Order Delete Message
struct OrderDeleteMessage{};
//! Order Replace Message
struct OrderReplaceMessage{};
//! Trade Message
struct TradeMessage{};
//! Cross Trade Message
struct CrossTradeMessage{};
//! Broken Trade Message
struct BrokenTradeMessage{};
//! Net Order Imbalance Indicator (NOII) Message
struct NOIIMessage{};
//! Retail Price Improvement Indicator (RPII) Message
struct RPIIMessage{};
//! Limit Up – Limit Down (LULD) Auction Collar Message
struct LULDAuctionCollarMessage{};
//! Unknown message
struct UnknownMessage{};
//! NASDAQ ITCH handler class
/*!
    NASDAQ ITCH handler is used to parse NASDAQ ITCH protocol and handle its
    messages in special handlers.

    NASDAQ ITCH protocol specification:
    http://www.nasdaqtrader.com/content/technicalsupport/specifications/dataproducts/NQTVITCHSpecification.pdf

    NASDAQ ITCH protocol examples:
    https://emi.nasdaq.com/ITCH

    Not thread-safe.
*/
class ITCHHandler
{
public:
    ITCHHandler() { Reset(); }
    ITCHHandler(const ITCHHandler&) = delete;
    ITCHHandler(ITCHHandler&&) = delete;
    virtual ~ITCHHandler() = default;

    ITCHHandler& operator=(const ITCHHandler&) = delete;
    ITCHHandler& operator=(ITCHHandler&&) = delete;
    bool Process(void* buffer, size_t size);
    bool ProcessMessage(void* buffer, size_t size);
    //! Reset ITCH handler
    void Reset();

protected:
    // Message handlers 虚函数 可以被子类重写
    virtual bool onMessage(const SystemEventMessage& message) { return true; }
    virtual bool onMessage(const StockDirectoryMessage& message) { return true; }
    virtual bool onMessage(const StockTradingActionMessage& message) { return true; }
    virtual bool onMessage(const RegSHOMessage& message) { return true; }
    virtual bool onMessage(const MarketParticipantPositionMessage& message) { return true; }
    virtual bool onMessage(const MWCBDeclineMessage& message) { return true; }
    virtual bool onMessage(const MWCBStatusMessage& message) { return true; }
    virtual bool onMessage(const IPOQuotingMessage& message) { return true; }
    virtual bool onMessage(const AddOrderMessage& message) { return true; }
    virtual bool onMessage(const AddOrderMPIDMessage& message) { return true; }
    virtual bool onMessage(const OrderExecutedMessage& message) { return true; }
    virtual bool onMessage(const OrderExecutedWithPriceMessage& message) { return true; }
    virtual bool onMessage(const OrderCancelMessage& message) { return true; }
    virtual bool onMessage(const OrderDeleteMessage& message) { return true; }
    virtual bool onMessage(const OrderReplaceMessage& message) { return true; }
    virtual bool onMessage(const TradeMessage& message) { return true; }
    virtual bool onMessage(const CrossTradeMessage& message) { return true; }
    virtual bool onMessage(const BrokenTradeMessage& message) { return true; }
    virtual bool onMessage(const NOIIMessage& message) { return true; }
    virtual bool onMessage(const RPIIMessage& message) { return true; }
    virtual bool onMessage(const LULDAuctionCollarMessage& message) { return true; }
    virtual bool onMessage(const UnknownMessage& message) { return true; }

private:
    // _size：当前正在处理的消息的总长度（0 表示等待新消息）
    size_t _size;
    // _cache：std::vector<uint8_t> 类型的缓存，用于存储不完整的消息
    std::vector<uint8_t> _cache;

    bool ProcessSystemEventMessage(void* buffer, size_t size);
    bool ProcessStockDirectoryMessage(void* buffer, size_t size);
    bool ProcessStockTradingActionMessage(void* buffer, size_t size);
    bool ProcessRegSHOMessage(void* buffer, size_t size);
    bool ProcessMarketParticipantPositionMessage(void* buffer, size_t size);
    bool ProcessMWCBDeclineMessage(void* buffer, size_t size);
    bool ProcessMWCBStatusMessage(void* buffer, size_t size);
    bool ProcessIPOQuotingMessage(void* buffer, size_t size);
    bool ProcessAddOrderMessage(void* buffer, size_t size);
    bool ProcessAddOrderMPIDMessage(void* buffer, size_t size);
    bool ProcessOrderExecutedMessage(void* buffer, size_t size);
    bool ProcessOrderExecutedWithPriceMessage(void* buffer, size_t size);
    bool ProcessOrderCancelMessage(void* buffer, size_t size);
    bool ProcessOrderDeleteMessage(void* buffer, size_t size);
    bool ProcessOrderReplaceMessage(void* buffer, size_t size);
    bool ProcessTradeMessage(void* buffer, size_t size);
    bool ProcessCrossTradeMessage(void* buffer, size_t size);
    bool ProcessBrokenTradeMessage(void* buffer, size_t size);
    bool ProcessNOIIMessage(void* buffer, size_t size);
    bool ProcessRPIIMessage(void* buffer, size_t size);
    bool ProcessLULDAuctionCollarMessage(void* buffer, size_t size);
    bool ProcessUnknownMessage(void* buffer, size_t size);

    template <size_t N>
    size_t ReadString(const void* buffer, char (&str)[N]);
    size_t ReadTimestamp(const void* buffer, uint64_t& value);
};

/*! \example itch_handler.cpp NASDAQ ITCH handler example */

} // namespace ITCH
} // namespace CppTrader

#include "itch_handler.inl"

#endif // CPPTRADER_ITCH_HANDLER_H

1、消息类型

ITCH协议中有不同类型的消息 ITCH协议消息类型 ,因此通过struct数据结构类定义不同类型的消息实体对象，例如SystemEventMessage表示系统事件消息;

2、协议处理器 ITCHHandler

协议处理器主要用来读取并且解析ITCH协议中不同的消息，其中设计了两个私有属性:_size表示当前处理消息的总长度，_cache:用于存储不完整的消息；

onMessage虚函数，不同用户可以对消息的处理可以灵活的自定义，通过重写onMessage函数，实现对消息的自定义处理，在本项目中，MyITCHHandler自定义处理器通过共有继承ITCHHandler处理器，然后重写onMessage函数来输出读取到的事件消息;

bool onMessage(const SystemEventMessage& message) override { 
    return OutputMessage(message); 
}

static bool OutputMessage(const TMessage& message)
 {
    std::cout << message << std::endl;
    return true;
}

Process* 协议处理器中，对ITCH协议中不同的消息，调用不同的方法进行解析处理，解析方法以Process*开头的方法;

在itch_hander.cpp文件中，对ITCH处理器的各种方法进行实现，应用中也实现了对处理器性能测试案例，请参考:ITCH处理器基准测试

Process 方法实现

/*!
    \file itch_handler.cpp
    \brief NASDAQ ITCH handler implementation
    \author Ivan Shynkarenka
    \date 21.07.2017
    \copyright MIT License
*/

#include "trader/providers/nasdaq/itch_handler.h"

#include <cassert>

namespace CppTrader {
namespace ITCH {

      /**
     * 功能：处理从网络接收的原始数据流，解析出完整的ITCH消息
     * 
     * ITCH协议特点：
     * - 每条消息以2字节长度字段开头（大端序）
     * - 消息是流式的，可能被拆分成多个TCP包传输
     * - 需要处理粘包和拆包问题
     * 
     * 状态管理：
     * - _size: 当前正在解析的消息总长度（0表示未开始解析新消息）
     * - _cache: 缓存不完整的消息数据
     */
bool ITCHHandler::Process(void* buffer, size_t size)
{
    /**
     * size_t index = 0;
        size_t：无符号整数类型，保证足够大以表示内存中任意对象的大小

        用途：作为游标，记录当前解析到数据流的哪个位置

        典型值：在 64 位系统上是 8 字节，在 32 位系统上是 4 字节
     */
    // 解析游标：记录当前已处理到的位置
    size_t index = 0;
    // 将输入缓冲区转换为字节指针，便于逐字节操作
    uint8_t* data = (uint8_t*)buffer;

    /**
     * ITCH 消息格式：每条消息以 2 字节长度字段开头（小端序）

        为什么是 3 字节？：需要至少 2 字节获取长度，加上可能需要缓存

        缓存状态：

        _cache.size() == 0 且 remaining < 3：数据不足，先缓存 1 字节

        _cache.size() == 1：已有 1 字节缓存，继续收集
     */

    /**
     * 主循环：处理输入缓冲区中的所有数据
     * 使用while循环而不是for，因为处理过程中可能跳过不定长度
     */
    while (index < size)
    {
         // ==================== 阶段1：读取消息长度 ====================
        // 当前没有正在解析的消息，需要读取新消息的长度字段
        if (_size == 0)
        {
            // 输入缓冲区中剩余未处理的数据量
            size_t remaining = size - index;

            /**
             * 情况A：长度字段不完整，需要缓存
             * 
             * 触发条件：
             * 1. _cache为空 且 剩余数据不足3字节（2字节长度+至少1字节数据）
             * 2. _cache已有1字节（说明上次收到了1字节，这次继续收集）
             * 
             * 为什么是3字节？因为最少需要2字节长度+1字节消息类型才能判断
             */
            // Collect message size into the cache
            if (((_cache.size() == 0) && (remaining < 3)) || (_cache.size() == 1))
            {
                _cache.push_back(data[index++]);
                continue;
            }

            /**
             * 情况B：有足够的数据读取消息长度
             */
            // Read a new message size
            uint16_t message_size;
            if (_cache.empty())
            {
                /**
                 * 子情况B1：没有缓存数据，直接从输入缓冲区读取
                 * ReadBigEndian: 将大端序的2字节转换为本机字节序
                 * 注意：这里index会在ReadBigEndian内部自增2
                 */
                // Read the message size directly from the input buffer
                index += CppCommon::Endian::ReadBigEndian(&data[index], message_size);
            }
            else
            {
                 /**
                 * 子情况B2：有缓存数据（说明之前收到了不完整的长度字段）
                 * 从缓存中读取长度字段，然后清空缓存
                 */
                // Read the message size from the cache
                CppCommon::Endian::ReadBigEndian(_cache.data(), message_size);

                // Clear the cache
                _cache.clear();
            }
            // 保存当前消息的总长度，进入下一阶段
            _size = message_size;
        }

         // ==================== 阶段2：读取消息体 ====================
        // 有正在解析的消息（_size > 0），需要收集完整的消息体
        // Read a new message
        if (_size > 0)
        {
            size_t remaining = size - index;// 剩余未处理的数据量

              /**
             * 情况A：消息体不完整，需要缓存
             */
            // Complete or place the message into the cache
            if (!_cache.empty())
            {
                /**
                 * 子情况A1：之前已有部分缓存（消息体被拆包了）
                 * tail = 还需要多少字节才能完成当前消息
                 * tail = 消息总长度 - 已缓存长度
                 */
                size_t tail = _size - _cache.size();
                if (tail > remaining)
                    tail = remaining;
                      // 将新数据追加到缓存末尾
                _cache.insert(_cache.end(), &data[index], &data[index + tail]);
                index += tail;
                // 如果缓存中的数据仍然不足完整消息，继续等待下次调用
                if (_cache.size() < _size)
                    continue;
            }
            /**
                 * 子情况A2：没有缓存，但剩余数据不够完整消息
                 * 这种情况通常发生在：刚好解析完上一条消息，新消息长度很大
                 * 当前TCP包只包含消息的开头部分
                 */
            else if (_size > remaining)
            {
                _cache.reserve(_size);// 预分配内存，避免多次扩容
                // 将剩余所有数据都放入缓存
                _cache.insert(_cache.end(), &data[index], &data[index + remaining]);
                index += remaining;
                continue;
            }
            /**
             * 情况B：有完整的消息可以处理
             */
            // Process the current message
            if (_cache.empty())
            {
                 /**
                 * 子情况B1：消息完整且在输入缓冲区中是连续的
                 * 直接处理输入缓冲区中的消息数据
                 */
                // Process the current message size directly from the input buffer
                if (!ProcessMessage(&data[index], _size))
                    return false;
                index += _size;
            }
            else
            {
                /**
                 * 子情况B2：消息数据来自缓存（之前不完整的数据）
                 * 处理缓存中的完整消息
                 */
                // Process the current message size directly from the cache
                if (!ProcessMessage(_cache.data(), _size))
                    return false;

                // Clear the cache                // 清空缓存，准备处理下一条消息

                _cache.clear();
            }
            /**
             * 重置消息长度，表示当前消息已处理完毕
             * 循环将继续，处理下一条消息
             */
            // Process the next message
            _size = 0;
        }
    }

    return true;
}

/**
 * 开始处理读取的数据
 */
bool ITCHHandler::ProcessMessage(void* buffer, size_t size)
{
    // Message is empty
    if (size == 0)
        return false;

    uint8_t* data = (uint8_t*)buffer;

    switch (*data)
    {
        // SystemEventMessage 消息类型为S
        case 'S': //系统事件消息，标识市场或数据源的启动、结束等状态
            return ProcessSystemEventMessage(data, size);
        // StockDirectoryMessage 消息类型为R
        case 'R': //股票目录消息，提供股票的基本信息和交易状态。 
            return ProcessStockDirectoryMessage(data, size);
        case 'H': //股票交易行动消息，指示某支股票的交易状态（如暂停、恢复）
            return ProcessStockTradingActionMessage(data, size);
        case 'Y': //Reg SHO限制消息，根据Reg SHO规则指示股票的卖空状态。
            return ProcessRegSHOMessage(data, size);
        case 'L'://市场参与者头寸消息，标识特定市场参与者在某股票上的持仓状态
            return ProcessMarketParticipantPositionMessage(data, size);
        case 'V'://市场-wide熔断机制（MWCB）下跌区间消息。
            return ProcessMWCBDeclineMessage(data, size);
        case 'W'://市场-wide熔断机制（MWCB）状态消息。
            return ProcessMWCBStatusMessage(data, size);
        case 'K'://IPO报价时段更新消息。   
            return ProcessIPOQuotingMessage(data, size);
        case 'A': //添加订单消息（无MPID归属），表示一个新的限价订单进入市场。
            return ProcessAddOrderMessage(data, size);
        case 'F'://添加订单消息（有MPID归属），与`A`类似，但包含了执行经纪商的ID
            return ProcessAddOrderMPIDMessage(data, size);
        case 'E'://订单执行消息，表示订单部分或全部成交。    
            return ProcessOrderExecutedMessage(data, size);
        case 'C'://订单以指定价格执行消息，用于成交价与原显示价不同的情况（如非显示订单执行）。
            return ProcessOrderExecutedWithPriceMessage(data, size);
        case 'X'://订单取消消息，表示订单被部分取消（余量更新）。
            return ProcessOrderCancelMessage(data, size);
        case 'D'://订单删除消息，表示订单被完全取消
            return ProcessOrderDeleteMessage(data, size);
        case 'U'://订单替换消息，表示一个订单被修改（如数量或价格变更）。
            return ProcessOrderReplaceMessage(data, size);
        case 'P'://非交叉交易消息，报告常规撮合产生的成交
            return ProcessTradeMessage(data, size);
        case 'Q'://交叉交易消息，报告开盘、收盘或IPO等集合竞价产生的成交。
            return ProcessCrossTradeMessage(data, size);
        case 'B'://交易作废消息，报告一笔之前公布的成交被作废。
            return ProcessBrokenTradeMessage(data, size);
        case 'I'://净订单 imbalance指标消息，提供开盘和收盘集合竞价前的 imbalance 信息。 |
            return ProcessNOIIMessage(data, size);
        case 'N': //零售价格改进指示器消息。 
            return ProcessRPIIMessage(data, size);
        case 'J': //涨跌幅限制（LULD）拍卖区间消息。 
            return ProcessLULDAuctionCollarMessage(data, size);
        default: //其他消息
            return ProcessUnknownMessage(data, size);
    }
}

void ITCHHandler::Reset()
{
    _size = 0;
    _cache.clear();
}

/**
 * 系统事件消息，标识市场或数据源的启动、结束等状态
 */
bool ITCHHandler::ProcessSystemEventMessage(void* buffer, size_t size)
{
    // ITCH 系统事件消息的固定长度为 12 字节，使用 assert 在调试模式下进行断言检查
    assert((size == 12) && "Invalid size of the ITCH message type 'S'");
    if (size != 12)
        return false;

    // 将 void* 转换为 uint8_t* 以便按字节操作
    uint8_t* data = (uint8_t*)buffer;

    SystemEventMessage message;
    // *data++ 先取当前指针的值，然后指针自增
    message.Type = *data++;
    // ITCH 协议使用大端序（网络字节序） ReadBigEndian 函数从网络字节序转换为主机字节序 返回读取的字节数（2），用于移动指针
    data += CppCommon::Endian::ReadBigEndian(data, message.StockLocate);
    data += CppCommon::Endian::ReadBigEndian(data, message.TrackingNumber);
    //读取 6 字节的纳秒级时间戳,返回读取的字节数（6）
    data += ReadTimestamp(data, message.Timestamp);
    message.EventCode = *data++;

    return onMessage(message);
}

} // namespace ITCH
} // namespace CppTrader

根据代码，ProcessSystemEventMessage护理方法，系统事件消息的字段布局如下：

字段	字节偏移	长度	说明
Type	0	1 字节	消息类型，应为 'S'
Stock Locate	1	2 字节	股票定位码
Tracking Number	3	2 字节	追踪编号
Timestamp	5	6 字节	时间戳（纳秒）
Event Code	11	1 字节	事件代码

根据 ITCH 规范，Event Code 可能的值包括：

代码	含义	说明
'O'	Start of Messages	消息开始
'S'	Start of System Hours	系统小时开始
'Q'	Start of Market Hours	市场小时开始
'M'	End of Market Hours	市场小时结束
'E'	End of System Hours	系统小时结束
'C'	End of Messages	消息结束

处理数据技术优化分析

ITCH处理器在基准测试中，吞吐量在 6489296 msg/s ，因此下面我们分析在process中使用了哪些技术来提高吞吐量;

1、零拷贝设计（Zero-Copy）

直接指针操作

uint8_t* data = (uint8_t*)buffer;  // 直接使用原始缓冲区，不复制

• 传统做法：将数据拷贝到内部缓冲区 → 额外的内存分配和复制
• CppTrader做法：直接在原始缓冲区上操作 → 减少50-80%的内存操作

为什么uint8_t* data = (uint8_t*)buffer操作是直接内存操作:零拷贝直接内存操作

条件性缓存

if (_cache.empty()) {
    // 路径1：无缓存，直接处理原始数据（零拷贝）
    ProcessMessage(&data[index], _size);
} else {
    // 路径2：有缓存才复制（仅在必要时）
    ProcessMessage(_cache.data(), _size);
}

关键优化：只在消息被拆包时才复制数据，理想情况下（数据完整到达）实现完全零拷贝。

2、分支预测优化（Branch Prediction）

常见路径优先

if (_cache.empty())  // 绝大多数情况为 true（无缓存）
{
    // 快速路径：直接处理
    ProcessMessage(&data[index], _size);
}
else  // 罕见情况（<1%）
{
    // 慢速路径：处理缓存
}

状态机设计

if (_size == 0) { /* 读取长度 */ }
if (_size > 0)  { /* 读取消息体 */ }

• 状态机让CPU分支预测器能够准确预测执行路径
• 在现代CPU上，分支预测错误惩罚约 10-20个时钟周期

3、内存访问模式优化

顺序访问

while (index < size) {
    // 按顺序访问 data[index], data[index+1], ...
    index += ReadBigEndian(&data[index], message_size);
    // ...
}

• 顺序内存访问能充分利用CPU的预取机制
• L1/L2/L3缓存的命中率可达 95%以上

缓存行对齐

uint8_t data[8192];  // 8KB，正好是L1缓存的典型大小

• 缓冲区大小与CPU缓存行大小对齐
• 减少跨缓存行访问

4、减少内存分配

预分配策略

_cache.reserve(_size);  // 预分配，避免多次扩容

• 每次扩容会导致 O(n)的复制开销
• 预分配将多次分配降为1次

缓存重用

_cache.clear();  // 保留已分配的内存

• clear() 不释放内存，只重置大小
• 下次使用时直接复用已有内存

5、批量处理

消息批处理

while (index < size)  // 一次调用处理多条消息

• 减少函数调用开销
• 利用CPU流水线并行处理

缓冲区批量读取

size = input->Read(buffer, sizeof(buffer));  // 一次读8KB

• 减少系统调用次数（从百万级降到数千次）

6、位操作优化

字节序转换

// ReadBigEndian 的实现（典型）
inline size_t ReadBigEndian(const uint8_t* data, uint16_t& value) {
    value = (data[0] << 8) | data[1];  // 位运算，极快
    return 2;
}

• 使用位运算替代乘法/除法
• 编译器会优化为单条指令（如 bswap）

为什么有这些优化

CPU架构特性

现代CPU（Intel/AMD）的特点：

• L1缓存: 32KB，延迟~4个周期
• L2缓存: 256KB，延迟~12个周期
• L3缓存: 8-32MB，延迟~40个周期
• 主内存: 延迟~200个周期

CppTrader的设计确保90%以上的数据访问在L1/L2缓存中完成。

指令流水线

; 典型处理流程（简化）
movzx eax, byte ptr [rdi]    ; 读1字节
shl   eax, 8                  ; 左移
movzx ecx, byte ptr [rdi+1]  ; 读2字节
or    eax, ecx                ; 合并
cmp   eax, 0                  ; 检查消息长度

• 流水线友好，没有分支中断

SIMD向量化

虽然这个特定代码没有显式使用SIMD，但编译器在 -O3 优化下会自动向量化部分操作：

_cache.insert(_cache.end(), &data[index], &data[index + tail]);

可能被优化为 memcpy，进而使用SIMD指令（如 rep movsb）。

小结

ITCHHandler::Process 的快，本质上来自于：

1. 架构设计：状态机 + 零拷贝 + 缓存优化
2. CPU友好：顺序访问 + 分支预测 + 缓存局部性
3. 内存管理：预分配 + 重用 + 避免复制
4. 批量处理：减少调用次数 + 充分利用流水线

这些优化让它在单核上达到接近内存带宽极限的性能，是高性能金融数据处理的一个典范实现。

Market_Manager处理器

市场管理主要用来管理订单簿的增删改查和用户订单的增删改查操作，MyMarketHandler 处理器继承自MarketHandler，MarketHandler 是一个抽象基类,用于接收和处理来自 MarketManager 的所有市场事件;数据结构设计可参考:MarketHandler 数据结构设计

MyMarketHandler自定义处理器

/**
 * 市场事件处理器
 * 继承自MarketHandler，监听MarketManager产生的所有市场事件
 * 用于统计各类市场操作的指标
 */
class MyMarketHandler : public MarketHandler
{
public:
 /**
     * 构造函数：初始化所有统计计数器为0
     */
    MyMarketHandler()
        : _updates(0),
          _symbols(0),
          _max_symbols(0),
          _order_books(0),
          _max_order_books(0),
          _max_order_book_levels(0),
          _max_order_book_orders(0),
          _orders(0),
          _max_orders(0),
          _add_orders(0),
          _update_orders(0),
          _delete_orders(0),
          _execute_orders(0)
    {}

    // 公共getter方法：提供统计数据的只读访问
    size_t updates() const { return _updates; }
    size_t max_symbols() const { return _max_symbols; }
    size_t max_order_books() const { return _max_order_books; }
    size_t max_order_book_levels() const { return _max_order_book_levels; }
    size_t max_order_book_orders() const { return _max_order_book_orders; }
    size_t max_orders() const { return _max_orders; }
    size_t add_orders() const { return _add_orders; }
    size_t update_orders() const { return _update_orders; }
    size_t delete_orders() const { return _delete_orders; }
    size_t execute_orders() const { return _execute_orders; }

protected:
    /**
     * 以下方法重写自MarketHandler，在MarketManager触发相应事件时被调用
     * 每个方法负责更新对应的统计指标
     */
    /** 添加股票时触发 */
    void onAddSymbol(const Symbol& symbol) override {
         ++_updates;  //// 总更新数+1
         ++_symbols; // 活跃股票数+1
         _max_symbols = std::max(_symbols, _max_symbols); // 更新历史最大值
    }
    /** 删除股票时触发 */
    void onDeleteSymbol(const Symbol& symbol) override { 
        ++_updates; 
        --_symbols; 
    }
    /** 添加订单簿时触发（通常与添加股票同时发生） */
    void onAddOrderBook(const OrderBook& order_book) override {
        ++_updates;
        ++_order_books; // 订单簿数量+1
        _max_order_books = std::max(_order_books, _max_order_books); 
    }
     /** 订单簿更新时触发（买卖盘变化） */
    void onUpdateOrderBook(const OrderBook& order_book, bool top) override {
          // 记录历史最大订单簿深度：比较买盘和卖盘的价格层级数，取较大值
        _max_order_book_levels = std::max(std::max(order_book.bids().size(), order_book.asks().size()), _max_order_book_levels); 
    }
    /** 删除订单簿时触发 */
    void onDeleteOrderBook(const OrderBook& order_book) override { 
        ++_updates; 
        --_order_books; 
    }
    /** 添加价格层级时触发（某个价格水平第一次出现订单） */
    void onAddLevel(const OrderBook& order_book, const Level& level, bool top) override { 
        ++_updates; 
    }
    /** 更新价格层级时触发（该价格水平的订单数量变化） */
    void onUpdateLevel(const OrderBook& order_book, const Level& level, bool top) override { 
        ++_updates;
         _max_order_book_orders = std::max(level.Orders, _max_order_book_orders); 
    }
    /** 删除价格层级时触发（该价格水平订单全部移除） */
    void onDeleteLevel(const OrderBook& order_book, const Level& level, bool top) override { 
        ++_updates; 
    }
    /** 添加订单时触发 */
    void onAddOrder(const Order& order) override { 
        ++_updates; 
        ++_orders; 
        _max_orders = std::max(_orders, _max_orders); ++_add_orders; 
    }
     /** 更新订单时触发（如价格或数量变更） */
    void onUpdateOrder(const Order& order) override { 
        ++_updates; 
        ++_update_orders; 
    }
    /** 删除订单时触发（完全取消） */
    void onDeleteOrder(const Order& order) override { 
        ++_updates; 
        --_orders; 
        ++_delete_orders; 
    }
    /** 订单成交时触发 */
    void onExecuteOrder(const Order& order, uint64_t price, uint64_t quantity) override { 
        ++_updates; 
        ++_execute_orders; 
    }

private:
    // 统计成员变量
    size_t _updates;                 // 市场事件总更新次数
    size_t _symbols;                 // 当前股票数量
    size_t _max_symbols;             // 历史最大股票数量
    size_t _order_books;             // 当前订单簿数量
    size_t _max_order_books;         // 历史最大订单簿数量
    size_t _max_order_book_levels;   // 历史最大订单簿深度
    size_t _max_order_book_orders;   // 历史单个价格层级最大订单数
    size_t _orders;                  // 当前未成交订单数
    size_t _max_orders;              // 历史最大未成交订单数
    size_t _add_orders;              // 添加订单操作总数
    size_t _update_orders;           // 更新订单操作总数
    size_t _delete_orders;           // 删除订单操作总数
    size_t _execute_orders;          // 成交订单操作总数
};

自定义市场处理器继承自抽象的MarketHandler 类，实现了对订单，符号，订单簿等操作的虚方法；

订单簿深度:

// 记录历史最大订单簿深度（买卖盘价格层级数的最大值）
_max_order_book_levels = std::max(
    std::max(order_book.bids().size(), order_book.asks().size()), 
    _max_order_book_levels
);

// 记录历史单个价格层级最大订单数量
_max_order_book_orders = std::max(level.Orders, _max_order_book_orders);

MyITCHHandler自定义处理器

MyITCHHandler处理器继承自抽象类ITCHHandler，实现对不同消息的重写操作;

/**
 * ITCH协议处理器
 * 继承自ITCHHandler，解析NASDAQ ITCH协议消息
 * 将解析后的消息转换为MarketManager的操作
 */
class MyITCHHandler : public ITCHHandler
{
public:
    /**
     * 构造函数
     * @param market 市场管理器引用，用于执行订单操作
     */
    MyITCHHandler(MarketManager& market)
        : _market(market),// 保存市场管理器引用
          _messages(0),// 已处理消息总数
          _errors(0)
    {}

    size_t messages() const { return _messages; }
    size_t errors() const { return _errors; }

protected:
/**
     * 以下方法重写自ITCHHandler，在解析到对应消息类型时被调用
     * 每个方法负责将ITCH消息转换为MarketManager的API调用
     */
     /** 系统事件消息（如市场开盘/收盘） */
    bool onMessage(const SystemEventMessage& message) override { 
        ++_messages; return true; // 仅计数，不产生市场操作
    }
    /** 股票目录消息 - 定义新股票 */
    bool onMessage(const StockDirectoryMessage& message) override {
         ++_messages; 
         Symbol symbol(message.StockLocate, message.Stock); // 创建Symbol对象（股票定位码，股票代码）
         _market.AddSymbol(symbol); // 向市场添加股票
         _market.AddOrderBook(symbol);// 为股票创建订单簿
        return true; 
    }
     /** 股票交易状态消息（如暂停/恢复交易） */
    bool onMessage(const StockTradingActionMessage& message) override { 
        ++_messages; 
        return true; 
    }
     /** REG SHO消息（卖空限制相关） */
    bool onMessage(const RegSHOMessage& message) override { 
        ++_messages; 
        return true; 
    }
    /** 市场参与者持仓消息 */
    bool onMessage(const MarketParticipantPositionMessage& message) override { ++_messages; return true; }
    /** MWCB（市场范围内熔断机制）下降消息 */
    bool onMessage(const MWCBDeclineMessage& message) override { ++_messages; return true; }
    bool onMessage(const MWCBStatusMessage& message) override { ++_messages; return true; }
    bool onMessage(const IPOQuotingMessage& message) override { ++_messages; return true; }
     /** 添加订单消息（无MPID） */
    bool onMessage(const AddOrderMessage& message) override { 
        ++_messages; 
        // 创建限价单并添加到市场
        // 参数：订单号、股票定位码、买卖方向、价格、数量
        _market.AddOrder(Order::Limit(message.OrderReferenceNumber, 
        message.StockLocate, 
        (message.BuySellIndicator == 'B') ? OrderSide::BUY : OrderSide::SELL, 
        message.Price, 
        message.Shares)); 
        return true;
    }
     /** 添加订单消息（包含MPID - 市场参与者ID） */
    bool onMessage(const AddOrderMPIDMessage& message) override { 
        ++_messages;
         _market.AddOrder(Order::Limit(message.OrderReferenceNumber, message.StockLocate, (message.BuySellIndicator == 'B') ? OrderSide::BUY : OrderSide::SELL, message.Price, message.Shares)); 
         return true; 
        }
    /** 订单成交消息（无成交价格，使用订单原价） */
    bool onMessage(const OrderExecutedMessage& message) override { 
        ++_messages; 
        _market.ExecuteOrder(message.OrderReferenceNumber, message.ExecutedShares); 
        return true; 
        }
     /** 订单成交消息（包含成交价格，可能用于跨市场交易） */
    bool onMessage(const OrderExecutedWithPriceMessage& message) override {
         ++_messages; 
         _market.ExecuteOrder(message.OrderReferenceNumber, message.ExecutionPrice, message.ExecutedShares); 
         return true; 
        }
    /** 订单取消消息（部分取消） */
    bool onMessage(const OrderCancelMessage& message) override { ++_messages; _market.ReduceOrder(message.OrderReferenceNumber, message.CanceledShares); return true; }
    /** 订单删除消息（完全取消） */
    bool onMessage(const OrderDeleteMessage& message) override { ++_messages; _market.DeleteOrder(message.OrderReferenceNumber); return true; }
    /** 订单替换消息（修改订单） */
    bool onMessage(const OrderReplaceMessage& message) override { ++_messages; _market.ReplaceOrder(message.OriginalOrderReferenceNumber, message.NewOrderReferenceNumber, message.Price, message.Shares); return true; }
    /** 成交消息（非订单驱动的成交） */
    bool onMessage(const TradeMessage& message) override { ++_messages; return true; }
    bool onMessage(const CrossTradeMessage& message) override { ++_messages; return true; }
    bool onMessage(const BrokenTradeMessage& message) override { ++_messages; return true; }
    bool onMessage(const NOIIMessage& message) override { ++_messages; return true; }
    bool onMessage(const RPIIMessage& message) override { ++_messages; return true; }
    bool onMessage(const LULDAuctionCollarMessage& message) override { ++_messages; return true; }
    bool onMessage(const UnknownMessage& message) override { ++_errors; return true; }

private:
    MarketManager& _market;// 市场管理器引用，用于执行订单操作
    size_t _messages;// 已处理消息总数
    size_t _errors;// 错误消息数量
};

主业务程序

/**
 * 主函数：程序入口
 * 功能：解析命令行参数，读取ITCH数据，执行基准测试，输出统计结果
 */
int main(int argc, char** argv)
{
    // 创建命令行参数解析器
    auto parser = optparse::OptionParser().version("1.0.0.0");

    // 添加命令行选项：-i 或 --input 用于指定输入文件
    parser.add_option("-i", "--input").dest("input").help("Input file name");
    // 解析命令行参数
    optparse::Values options = parser.parse_args(argc, argv);

    // Print help // 如果请求帮助信息，打印帮助并退出
    if (options.get("help"))
    {
        parser.print_help();
        return 0;
    }
    // 创建市场事件处理器和市场管理器
    MyMarketHandler market_handler;// 统计处理器
    MarketManager market(market_handler);// 市场管理器（传入处理器以接收事件通知）
    MyITCHHandler itch_handler(market);// ITCH处理器（传入市场管理器以执行操作）

    // 配置输入源：默认从标准输入读取
    // Open the input file or stdin
    std::unique_ptr<Reader> input(new StdInput());
    if (options.is_set("input"))
    {
        File* file = new File(Path(options.get("input")));
        file->Open(true, false);
        input.reset(file);
    }

    // Perform input
    size_t size;
    uint8_t buffer[8192]; // 8KB缓冲区，平衡内存使用和系统调用次数
    std::cout << "ITCH processing...";
    uint64_t timestamp_start = Timestamp::nano(); // 记录开始时间（纳秒）
    // 循环读取数据直到文件结束
    while ((size = input->Read(buffer, sizeof(buffer))) > 0)
    {
        // 处理缓冲区中的数据：解析ITCH消息并更新市场状态
        // Process the buffer
        itch_handler.Process(buffer, size);
    }
    uint64_t timestamp_stop = Timestamp::nano();
    std::cout << "Done!" << std::endl;

    std::cout << std::endl;

    std::cout << "Errors: " << itch_handler.errors() << std::endl;

    std::cout << std::endl;

    size_t total_messages = itch_handler.messages();
    size_t total_updates = market_handler.updates();

    std::cout << "Processing time: " << CppBenchmark::ReporterConsole::GenerateTimePeriod(timestamp_stop - timestamp_start) << std::endl;
    std::cout << "Total ITCH messages: " << total_messages << std::endl;
    std::cout << "ITCH message latency: " << CppBenchmark::ReporterConsole::GenerateTimePeriod((timestamp_stop - timestamp_start) / total_messages) << std::endl;
    std::cout << "ITCH message throughput: " << total_messages * 1000000000 / (timestamp_stop - timestamp_start) << " msg/s" << std::endl;
    std::cout << "Total market updates: " << total_updates << std::endl;
    std::cout << "Market update latency: " << CppBenchmark::ReporterConsole::GenerateTimePeriod((timestamp_stop - timestamp_start) / total_updates) << std::endl;
    std::cout << "Market update throughput: " << total_updates * 1000000000 / (timestamp_stop - timestamp_start) << " upd/s" << std::endl;

    std::cout << std::endl;

    std::cout << "Market statistics: " << std::endl;
    std::cout << "Max symbols: " << market_handler.max_symbols() << std::endl;
    std::cout << "Max order books: " << market_handler.max_order_books() << std::endl;
    std::cout << "Max order book levels: " << market_handler.max_order_book_levels() << std::endl;
    std::cout << "Max order book orders: " << market_handler.max_order_book_orders() << std::endl;
    std::cout << "Max orders: " << market_handler.max_orders() << std::endl;

    std::cout << std::endl;

    std::cout << "Order statistics: " << std::endl;
    std::cout << "Add order operations: " << market_handler.add_orders() << std::endl;
    std::cout << "Update order operations: " << market_handler.update_orders() << std::endl;
    std::cout << "Delete order operations: " << market_handler.delete_orders() << std::endl;
    std::cout << "Execute order operations: " << market_handler.execute_orders() << std::endl;

    return 0;
}

数据流:

ITCH二进制文件 → 缓冲区 → ITCH消息解析 → MarketManager API调用 → 市场状态更新 → 事件统计

Market manager (optimized version)

上一个版本的市场管理器是未经过优化的 未优化市场管理器，因此，在这个版本中，做如下优化:

1. 符号和订单簿存储在固定大小的预先分配的数组中.
2. 订单存储在预先分配的数组中，而非使用 HashMap。这使得所有订单操作（获取、插入、更新、删除）的执行时间均为 O(1)，且没有额外开销。
3. 没有维护价格级别的订单链表，而是仅记录订单数量。
4. 价格级别存储在排序数组中，而非红黑树中。排序顺序将最优价格（最优买入价/最优卖出价）置于数组末尾，这提供了良好的 CPU 缓存局部性，并且对于接近市场价格的订单具有接近 O(1) 的搜索时间，但对于远离市场价格的订单则会有所损失！
5. 价格级别是从池中获取的，该池是使用预先分配的数组实现的，每个价格级别的创建和删除操作的时间复杂度均为 O(1)。

Market manager (aggressive optimized version)

这是一个针对市场管理功能进行深度优化的版本。它能实现接近最优的延迟和吞吐量效果，但为此需要采用一些额外的优化手段，而这些手段在实际的交易平台上可能难以维持：

1. 符号未被保留
2. 订单和价格限制结构经过优化以达到最佳状态。大部分有用字段已被删除。
3. 价格值以有符号的 32 位整数形式存储。投标价格为正数，出价价格为负数。
4. 不再使用市场处理程序。无法从市场管理器接收通知。

matching_engine

ITCH processing...Done!

Errors: 0

Processing time: 5:23.394 m
Total ITCH messages: 368366634
ITCH message latency: 877 ns
ITCH message throughput: 1139063 msg/s
Total market updates: 789034636
Market update latency: 409 ns
Market update throughput: 2439853 upd/s

Market statistics:
Max symbols: 8713
Max order books: 8713
Max order book levels: 3421
Max order book orders: 3049
Max orders: 1922049

Order statistics:
Add order operations: 191500448
Update order operations: 7604314
Delete order operations: 191300543
Execute order operations: 17142148

贡献者

codingLab