业务延迟优化全栈指南：从硬件到应用的系统性降延迟实践

# 共置服务架构设计
colocation_strategy:
  # 核心撮合引擎与交易参与者的共置
  matching_engine:
    location: "同一数据中心"
    network_topology: "同一机架或相邻机架"
    latency_target: "< 100μs"
  
  # 数据库与应用的共置
  database:
    location: "与应用服务器同一机房"
    network_hops: 0  # 零跳转
    latency_target: "< 50μs"
  
  # 缓存层共置
  cache:
    location: "与应用进程同一主机"
    access_method: "本地内存或共享内存"
    latency_target: "< 1μs"

# CPU性能评估脚本
#!/bin/bash
# cpu_performance_check.sh

check_cpu_performance() {
    echo "=== CPU性能评估 ==="
    
    # 检查CPU主频
    CPU_FREQ=$(lscpu | grep "CPU max MHz" | awk '{print $4}')
    echo "最大主频: ${CPU_FREQ} MHz"
    
    # 检查L3缓存大小
    L3_CACHE=$(lscpu | grep "L3 cache" | awk '{print $3, $4}')
    echo "L3缓存: $L3_CACHE"
    
    # 检查CPU核心数
    CPU_CORES=$(lscpu | grep "^CPU(s):" | awk '{print $2}')
    echo "CPU核心数: $CPU_CORES"
    
    # 检查是否支持超线程
    THREADS_PER_CORE=$(lscpu | grep "Thread(s) per core" | awk '{print $4}')
    echo "每核心线程数: $THREADS_PER_CORE"
    
    # 推荐配置
    if (( $(echo "$CPU_FREQ > 4000" | bc -l) )); then
        echo "✓ 主频满足高频要求"
    else
        echo "✗ 建议选择主频 > 4.0GHz 的CPU"
    fi
}

check_cpu_performance

# CPU性能模式配置
#!/bin/bash
# cpu_optimization.sh

# 1. 设置CPU性能模式（禁用节能）
for cpu in /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor; do
    echo performance > $cpu
done

# 2. 禁用CPU节能特性（C-States）
# 在GRUB配置中添加
# GRUB_CMDLINE_LINUX="intel_idle.max_cstate=0 processor.max_cstate=0"

# 3. 禁用超线程（对于单线程关键路径）
# 在BIOS中禁用，或通过内核参数
# GRUB_CMDLINE_LINUX="nosmt"

# 4. 禁用Turbo Boost（可选，保持稳定频率）
echo 1 > /sys/devices/system/cpu/intel_pstate/no_turbo

# 5. 固定CPU频率
echo 5000000 > /sys/devices/system/cpu/cpu0/cpufreq/scaling_max_freq
echo 5000000 > /sys/devices/system/cpu/cpu0/cpufreq/scaling_min_freq

# 验证配置
echo "=== CPU配置验证 ==="
cat /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor
cat /sys/devices/system/cpu/cpu*/cpufreq/scaling_cur_freq

// FPGA网络协议栈卸载示例（伪代码）
module network_offload (
    input wire clk,
    input wire rst,
    input wire [63:0] packet_data,
    output reg [63:0] processed_data
);
    
    // UDP/TCP协议解析
    always @(posedge clk) begin
        if (packet_data[15:0] == 16'h0800) begin  // IPv4
            // 解析IP头
            // 解析TCP/UDP头
            // 提取应用数据
            processed_data <= extract_payload(packet_data);
        end
    end
endmodule

# FPGA加速架构
fpga_acceleration:
  network_stack:
    - udp_offload: "UDP协议栈硬件解析"
    - tcp_offload: "TCP协议栈硬件解析"
    - checksum_offload: "校验和硬件计算"
  
  business_logic:
    - risk_control: "风控规则硬件执行"
    - order_matching: "订单撮合逻辑硬件加速"
    - market_data: "行情数据处理硬件加速"
  
  performance_gain:
    latency_reduction: "90-95%"
    throughput_increase: "10-100x"

# 内存性能检查
#!/bin/bash
# memory_performance_check.sh

check_memory_performance() {
    echo "=== 内存性能评估 ==="
    
    # 检查内存类型和频率
    dmidecode -t memory | grep -E "Speed|Type|Size"
    
    # 检查内存延迟（使用工具如mlc）
    # mlc --latency_matrix
    
    # 检查NUMA拓扑
    numactl --hardware
    
    # 检查内存带宽
    # stream benchmark
}

# 内存优化配置
optimize_memory() {
    # 1. 启用大页内存
    echo 1024 > /proc/sys/vm/nr_hugepages
    echo 'vm.nr_hugepages = 1024' >> /etc/sysctl.conf
    
    # 2. NUMA绑定
    # 将进程绑定到特定NUMA节点
    numactl --membind=0 --cpunodebind=0 your_application
    
    # 3. 内存预分配
    # 在应用启动时预分配所有需要的内存
}

// DPDK数据包处理示例
#include <rte_eal.h>
#include <rte_ethdev.h>
#include <rte_mbuf.h>

int main(int argc, char *argv[]) {
    // 初始化DPDK环境
    rte_eal_init(argc, argv);
    
    // 配置网卡
    uint16_t port_id = 0;
    struct rte_eth_conf port_conf = {
        .rxmode = {
            .max_rx_pkt_len = RTE_ETHER_MAX_LEN,
            .offloads = DEV_RX_OFFLOAD_CHECKSUM,
        },
        .txmode = {
            .offloads = DEV_TX_OFFLOAD_IPV4_CKSUM | 
                       DEV_TX_OFFLOAD_UDP_CKSUM,
        },
    };
    
    rte_eth_dev_configure(port_id, 1, 1, &port_conf);
    
    // 分配接收队列
    struct rte_mempool *mbuf_pool = rte_pktmbuf_pool_create(
        "MBUF_POOL", 8192, 256, 0,
        RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id()
    );
    
    rte_eth_rx_queue_setup(port_id, 0, 512,
                           rte_eth_dev_socket_id(port_id),
                           NULL, mbuf_pool);
    
    // 启动网卡
    rte_eth_dev_start(port_id);
    
    // 数据包处理循环
    struct rte_mbuf *bufs[32];
    while (1) {
        uint16_t nb_rx = rte_eth_rx_burst(port_id, 0, bufs, 32);
        if (nb_rx > 0) {
            // 处理数据包（零拷贝）
            process_packets(bufs, nb_rx);
            // 发送响应
            rte_eth_tx_burst(port_id, 0, bufs, nb_rx);
        }
    }
    
    return 0;
}

# Solarflare Onload配置
# 1. 安装Onload
# 2. 配置环境变量
export LD_PRELOAD=/usr/lib64/libonload.so

# 3. 运行应用（自动使用Onload）
./your_application

# 4. 验证Onload状态
onload_stackdump lots

# RoCE v2网络配置
#!/bin/bash
# roce_configuration.sh

# 1. 检查RDMA设备
rdma dev show

# 2. 配置无损以太网（Lossless Ethernet）
# 交换机配置（以Cisco为例）
# interface Ethernet1/1
#   priority-flow-control mode on
#   priority-flow-control priority 3,4
#   ecn
#   ecn threshold 1000 10000

# 3. 主机端配置
# 启用PFC
echo 1 > /sys/class/net/ens1f0/device/sriov_numvfs
echo 0 > /sys/class/net/ens1f0/device/sriov_numvfs

# 配置网卡参数
ethtool -L ens1f0 combined 16
ethtool -G ens1f0 rx 4096 tx 4096
ethtool -K ens1f0 gro off lro off tso off gso off

# 4. 测试RDMA性能
ibv_rc_pingpong -d mlx5_0 -g 0 -s 64 -n 1000

// RDMA通信示例
#include <infiniband/verbs.h>

struct ibv_context *ctx;
struct ibv_pd *pd;
struct ibv_cq *cq;
struct ibv_qp *qp;

// 初始化RDMA资源
void init_rdma() {
    // 打开设备
    struct ibv_device **dev_list = ibv_get_device_list(NULL);
    ctx = ibv_open_device(dev_list[0]);
    
    // 创建保护域
    pd = ibv_alloc_pd(ctx);
    
    // 创建完成队列
    cq = ibv_create_cq(ctx, 10, NULL, NULL, 0);
    
    // 创建队列对
    struct ibv_qp_init_attr qp_init_attr = {
        .send_cq = cq,
        .recv_cq = cq,
        .cap = {
            .max_send_wr = 1024,
            .max_recv_wr = 1024,
            .max_send_sge = 16,
            .max_recv_sge = 16,
        },
        .qp_type = IBV_QPT_RC,
    };
    qp = ibv_create_qp(pd, &qp_init_attr);
}

// RDMA写操作（零拷贝）
void rdma_write(void *local_addr, uint32_t length, 
                uint32_t remote_addr, uint32_t rkey) {
    struct ibv_sge sge = {
        .addr = (uintptr_t)local_addr,
        .length = length,
        .lkey = mr->lkey,
    };
    
    struct ibv_send_wr wr = {
        .wr_id = 1,
        .sg_list = &sge,
        .num_sge = 1,
        .opcode = IBV_WR_RDMA_WRITE,
        .send_flags = IBV_SEND_SIGNALED,
        .wr = {
            .rdma = {
                .remote_addr = remote_addr,
                .rkey = rkey,
            },
        },
    };
    
    struct ibv_send_wr *bad_wr;
    ibv_post_send(qp, &wr, &bad_wr);
}

// UDP组播发送示例
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

int create_multicast_sender(const char *multicast_ip, int port) {
    int sock = socket(AF_INET, SOCK_DGRAM, 0);
    
    // 设置组播TTL
    int ttl = 1;  // 本地网络
    setsockopt(sock, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl));
    
    // 设置组播接口
    struct in_addr interface;
    interface.s_addr = INADDR_ANY;
    setsockopt(sock, IPPROTO_IP, IP_MULTICAST_IF, &interface, sizeof(interface));
    
    // 设置发送缓冲区
    int send_buf_size = 1024 * 1024;  // 1MB
    setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &send_buf_size, sizeof(send_buf_size));
    
    // 配置目标地址
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = inet_addr(multicast_ip);
    addr.sin_port = htons(port);
    
    // 发送数据
    // sendto(sock, data, len, 0, (struct sockaddr*)&addr, sizeof(addr));
    
    return sock;
}

// UDP组播接收示例
int create_multicast_receiver(const char *multicast_ip, int port) {
    int sock = socket(AF_INET, SOCK_DGRAM, 0);
    
    // 设置地址重用
    int reuse = 1;
    setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
    
    // 绑定地址
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = INADDR_ANY;
    addr.sin_port = htons(port);
    bind(sock, (struct sockaddr*)&addr, sizeof(addr));
    
    // 加入组播组
    struct ip_mreq mreq;
    mreq.imr_multiaddr.s_addr = inet_addr(multicast_ip);
    mreq.imr_interface.s_addr = INADDR_ANY;
    setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
    
    // 设置接收缓冲区
    int recv_buf_size = 1024 * 1024;  // 1MB
    setsockopt(sock, SOL_SOCKET, SO_RCVBUF, &recv_buf_size, sizeof(recv_buf_size));
    
    return sock;
}

# 交换机组播配置（以Cisco为例）
# interface Ethernet1/1
#   ip igmp version 3
#   ip pim sparse-mode
#   ip igmp static-group 239.1.1.1

# 主机端IGMP配置
# 确保IGMP协议正常工作
echo 1 > /proc/sys/net/ipv4/ip_forward
echo 1 > /proc/sys/net/ipv4/conf/all/mc_forwarding

# Leaf-Spine网络拓扑设计
network_topology:
  architecture: "Leaf-Spine"
  
  spine_layer:
    switches: 4
    ports_per_switch: 64
    bandwidth: "100Gbps per port"
    oversubscription: "1:1"  # 无阻塞设计
  
  leaf_layer:
    switches: 8
    ports_per_switch: 48
    server_ports: 32
    uplink_ports: 16
    bandwidth: "25Gbps per server port"
  
  routing:
    protocol: "ECMP"  # 等价多路径
    load_balancing: "5-tuple hash"
    failover_time: "< 50ms"
  
  latency_targets:
    same_rack: "< 5μs"
    same_leaf: "< 10μs"
    cross_leaf: "< 20μs"

# 网络路径优化脚本
#!/bin/bash
# network_path_optimization.sh

# 1. 配置静态路由（减少路由查找时间）
ip route add 10.0.0.0/8 via 10.0.1.1 dev eth0

# 2. 配置ECMP（等价多路径）
ip route add default \
    nexthop via 10.0.1.1 dev eth0 weight 1 \
    nexthop via 10.0.1.2 dev eth0 weight 1

# 3. 优化ARP缓存
echo 300 > /proc/sys/net/ipv4/neigh/default/gc_stale_time
echo 600 > /proc/sys/net/ipv4/neigh/default/base_reachable_time

# 4. 禁用ICMP重定向
echo 0 > /proc/sys/net/ipv4/conf/all/accept_redirects
echo 0 > /proc/sys/net/ipv4/conf/all/send_redirects

# CPU隔离配置脚本
#!/bin/bash
# cpu_isolation.sh

# 1. 在GRUB配置中隔离CPU核心
# GRUB_CMDLINE_LINUX="isolcpus=2,3,4,5 nohz_full=2,3,4,5 rcu_nocbs=2,3,4,5"

# 2. 配置CPU亲和性
isolate_cpus() {
    local cpus=$1
    local pid=$2
    
    # 使用taskset绑定进程到特定CPU
    taskset -cp $cpus $pid
    
    # 或使用cpuset
    echo $cpus > /sys/fs/cgroup/cpuset/cpuset.cpus
    echo $pid > /sys/fs/cgroup/cpuset/tasks
}

# 3. 禁用CPU调度（实时优先级）
chrt -f 99 your_application

# 4. 设置进程优先级
renice -20 -p $PID

# 验证CPU隔离效果
#!/bin/bash
# verify_cpu_isolation.sh

echo "=== CPU隔离验证 ==="

# 检查隔离的CPU
ISOLATED_CPUS=$(cat /proc/cmdline | grep -oP 'isolcpus=\K[0-9,]+')
echo "隔离的CPU: $ISOLATED_CPUS"

# 检查进程CPU绑定
ps -eo pid,psr,comm | grep your_application

# 检查CPU使用率
mpstat -P ALL 1 5

# 检查中断分布
cat /proc/interrupts | grep -E "CPU|mlx5"

# 中断亲和性配置脚本
#!/bin/bash
# irq_affinity.sh

# 1. 禁用中断均衡
systemctl stop irqbalance
systemctl disable irqbalance

# 2. 获取网卡中断号
get_irq_numbers() {
    local interface=$1
    local irq_file="/proc/interrupts"
    
    # 查找网卡对应的中断
    grep $interface $irq_file | awk '{print $1}' | sed 's/://'
}

# 3. 绑定中断到特定CPU（避免工作CPU）
bind_irq_to_cpu() {
    local irq=$1
    local cpu=$2
    
    # 设置中断亲和性（CPU掩码）
    local cpu_mask=$(echo "2^$cpu" | bc)
    printf "%x" $cpu_mask > /proc/irq/$irq/smp_affinity
}

# 4. 配置多队列网卡中断
configure_multiqueue_irq() {
    local interface="ens1f0"
    local num_queues=16
    local work_cpus="2,3,4,5"
    local irq_cpus="0,1"
    
    # 获取所有队列的中断号
    for queue in $(seq 0 $((num_queues-1))); do
        irq=$(get_irq_numbers "${interface}-TxRx-${queue}")
        if [ -n "$irq" ]; then
            # 将中断绑定到非工作CPU
            bind_irq_to_cpu $irq 0  # 绑定到CPU 0
        fi
    done
}

# 5. 验证中断分布
verify_irq_distribution() {
    echo "=== 中断分布 ==="
    watch -n 1 'cat /proc/interrupts | head -20'
}

# 中断优化配置
irq_optimization:
  # 工作CPU：运行关键业务逻辑
  work_cpus: [2, 3, 4, 5]
  
  # 中断CPU：处理网络中断
  irq_cpus: [0, 1]
  
  # 隔离CPU：完全隔离，只运行关键进程
  isolated_cpus: [6, 7]
  
  # 中断处理策略
  interrupt_handling:
    - disable_irqbalance: true
    - manual_affinity: true
    - use_irq_cpus: true
    - avoid_work_cpus: true

# 大页内存配置脚本
#!/bin/bash
# hugepages_config.sh

# 1. 检查当前大页配置
check_hugepages() {
    echo "=== 大页内存状态 ==="
    cat /proc/meminfo | grep -i huge
    cat /proc/sys/vm/nr_hugepages
}

# 2. 配置2MB大页
configure_2mb_hugepages() {
    local num_pages=$1  # 例如 1024 (2GB)
    
    # 临时配置
    echo $num_pages > /proc/sys/vm/nr_hugepages
    
    # 永久配置
    echo "vm.nr_hugepages = $num_pages" >> /etc/sysctl.conf
    
    # 验证
    cat /proc/sys/vm/nr_hugepages
}

# 3. 配置1GB大页（需要CPU支持）
configure_1gb_hugepages() {
    local num_pages=$1  # 例如 4 (4GB)
    
    # 检查CPU支持
    if grep -q pdpe1gb /proc/cpuinfo; then
        echo "CPU支持1GB大页"
        
        # 配置1GB大页
        echo $num_pages > /sys/kernel/mm/hugepages/hugepages-1048576kB/nr_hugepages
        
        # 永久配置
        echo "vm.nr_hugepages_1gb = $num_pages" >> /etc/sysctl.conf
    else
        echo "CPU不支持1GB大页"
    fi
}

# 4. 应用使用大页内存
use_hugepages_in_app() {
    # C/C++应用中使用mmap
    # void *ptr = mmap(NULL, size, PROT_READ|PROT_WRITE,
    #                  MAP_PRIVATE|MAP_ANONYMOUS|MAP_HUGETLB, -1, 0);
    
    # Java应用中使用-XX:+UseLargePages
    # java -XX:+UseLargePages YourApplication
}

# 5. 验证大页使用情况
verify_hugepages_usage() {
    echo "=== 大页使用情况 ==="
    cat /proc/meminfo | grep -i huge
    cat /sys/kernel/mm/hugepages/hugepages-2048kB/nr_hugepages
    cat /sys/kernel/mm/hugepages/hugepages-2048kB/free_hugepages
}

# 内核参数优化脚本
#!/bin/bash
# kernel_optimization.sh

configure_kernel_params() {
    cat >> /etc/sysctl.conf << 'EOF'
# 网络优化
net.core.rmem_max = 134217728          # 128MB
net.core.wmem_max = 134217728          # 128MB
net.core.rmem_default = 262144         # 256KB
net.core.wmem_default = 262144         # 256KB
net.core.netdev_max_backlog = 5000
net.core.netdev_budget = 600
net.core.somaxconn = 65535

# TCP优化
net.ipv4.tcp_rmem = 4096 262144 134217728
net.ipv4.tcp_wmem = 4096 262144 134217728
net.ipv4.tcp_congestion_control = bbr
net.ipv4.tcp_slow_start_after_idle = 0
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 30
net.ipv4.tcp_keepalive_time = 600
net.ipv4.tcp_keepalive_probes = 3
net.ipv4.tcp_keepalive_intvl = 15

# 内存管理优化
vm.swappiness = 10
vm.dirty_ratio = 15
vm.dirty_background_ratio = 5
vm.dirty_expire_centisecs = 3000
vm.dirty_writeback_centisecs = 500
vm.overcommit_memory = 1

# 文件系统优化
fs.file-max = 2097152
fs.nr_open = 2097152

# 进程和线程优化
kernel.pid_max = 4194304
kernel.threads-max = 2097152

# 禁用透明大页（THP）对延迟的影响
vm.transparent_hugepage = never

# NUMA优化
kernel.numa_balancing = 0
EOF

    # 应用配置
    sysctl -p
    
    # 验证配置
    sysctl -a | grep -E "net.core|net.ipv4.tcp|vm\."
}

# 禁用透明大页
disable_transparent_hugepages() {
    echo never > /sys/kernel/mm/transparent_hugepage/enabled
    echo never > /sys/kernel/mm/transparent_hugepage/defrag
    
    # 永久配置
    cat >> /etc/rc.local << 'EOF'
echo never > /sys/kernel/mm/transparent_hugepage/enabled
echo never > /sys/kernel/mm/transparent_hugepage/defrag
EOF
}

# 实时内核配置
# 1. 安装实时内核（RT Kernel）
# yum install kernel-rt
# 或
# apt install linux-image-rt-amd64

# 2. 配置实时调度策略
configure_realtime_scheduling() {
    local pid=$1
    local priority=99  # 最高实时优先级
    
    # 设置FIFO实时调度策略
    chrt -f $priority $pid
    
    # 或使用SCHED_DEADLINE（Linux 3.14+）
    # chrt -d --sched-runtime 1000000 \
    #        --sched-deadline 2000000 \
    #        --sched-period 2000000 \
    #        $pid
}

# 3. 配置CPU带宽控制（CGROUP）
configure_cpu_bandwidth() {
    # 创建cgroup
    mkdir -p /sys/fs/cgroup/cpu/realtime
    
    # 设置CPU带宽限制（例如50%）
    echo "50000 100000" > /sys/fs/cgroup/cpu/realtime/cpu.cfs_quota_us
    
    # 将进程加入cgroup
    echo $PID > /sys/fs/cgroup/cpu/realtime/cgroup.procs
}

// 内存撮合引擎示例
#include <unordered_map>
#include <queue>
#include <atomic>
#include <thread>

class InMemoryMatchingEngine {
private:
    // 订单簿（完全在内存中）
    struct OrderBook {
        std::map<double, std::queue<Order>> bids;  // 买单
        std::map<double, std::queue<Order>> asks;  // 卖单
    };
    
    OrderBook orderbook_;
    std::atomic<uint64_t> sequence_{0};
    
    // 异步持久化队列
    class AsyncPersistence {
    private:
        std::queue<OrderEvent> event_queue_;
        std::thread persistence_thread_;
        std::atomic<bool> running_{true};
        
    public:
        void persist_async(const OrderEvent& event) {
            event_queue_.push(event);
        }
        
        void start() {
            persistence_thread_ = std::thread([this]() {
                while (running_) {
                    if (!event_queue_.empty()) {
                        OrderEvent event = event_queue_.front();
                        event_queue_.pop();
                        // 写入顺序日志（AOF）
                        write_to_aof(event);
                    }
                    std::this_thread::sleep_for(
                        std::chrono::microseconds(100)
                    );
                }
            });
        }
    };
    
    AsyncPersistence persistence_;
    
public:
    // 撮合逻辑（完全在内存中，无I/O阻塞）
    MatchResult match_order(const Order& order) {
        auto start = std::chrono::high_resolution_clock::now();
        
        // 内存撮合逻辑
        MatchResult result = do_match(order);
        
        // 异步持久化（不阻塞撮合）
        OrderEvent event = create_event(order, result);
        persistence_.persist_async(event);
        
        auto end = std::chrono::high_resolution_clock::now();
        auto latency = std::chrono::duration_cast<
            std::chrono::microseconds>(end - start).count();
        
        // 记录延迟指标
        record_latency(latency);
        
        return result;
    }
};

// 顺序日志写入（高吞吐、低延迟）
class SequentialLogWriter {
private:
    int fd_;
    char* buffer_;
    size_t buffer_size_;
    size_t buffer_pos_;
    
public:
    SequentialLogWriter(const std::string& log_file) {
        // 使用O_DIRECT标志，绕过页缓存
        fd_ = open(log_file.c_str(), 
                   O_WRONLY | O_CREAT | O_APPEND | O_DIRECT, 
                   0644);
        
        // 分配对齐的缓冲区（O_DIRECT要求）
        buffer_size_ = 1024 * 1024;  // 1MB
        posix_memalign((void**)&buffer_, 512, buffer_size_);
        buffer_pos_ = 0;
    }
    
    void write_event(const OrderEvent& event) {
        // 序列化事件
        size_t event_size = serialize(event, 
                                     buffer_ + buffer_pos_,
                                     buffer_size_ - buffer_pos_);
        
        buffer_pos_ += event_size;
        
        // 缓冲区满时刷新
        if (buffer_pos_ >= buffer_size_ - 1024) {
            flush();
        }
    }
    
    void flush() {
        if (buffer_pos_ > 0) {
            // 直接写入磁盘（O_DIRECT）
            ssize_t written = write(fd_, buffer_, buffer_pos_);
            buffer_pos_ = 0;
        }
    }
};

// 基于RingBuffer的无锁队列（Disruptor模式）
#include <atomic>
#include <cstdint>

template<typename T, size_t Size>
class LockFreeRingBuffer {
private:
    static_assert((Size & (Size - 1)) == 0, "Size must be power of 2");
    
    T buffer_[Size];
    std::atomic<uint64_t> write_pos_{0};
    std::atomic<uint64_t> read_pos_{0};
    
    // 缓存行对齐，避免False Sharing
    alignas(64) std::atomic<uint64_t> cached_read_pos_{0};
    alignas(64) std::atomic<uint64_t> cached_write_pos_{0};
    
public:
    bool try_push(const T& item) {
        uint64_t current_write = write_pos_.load(std::memory_order_relaxed);
        uint64_t next_write = current_write + 1;
        
        // 检查队列是否满
        uint64_t cached_read = cached_read_pos_.load(std::memory_order_acquire);
        if (next_write - cached_read >= Size) {
            // 更新缓存的读位置
            cached_read = read_pos_.load(std::memory_order_acquire);
            cached_read_pos_.store(cached_read, std::memory_order_relaxed);
            
            if (next_write - cached_read >= Size) {
                return false;  // 队列满
            }
        }
        
        // 写入数据
        buffer_[current_write & (Size - 1)] = item;
        
        // 更新写位置（发布）
        write_pos_.store(next_write, std::memory_order_release);
        
        return true;
    }
    
    bool try_pop(T& item) {
        uint64_t current_read = read_pos_.load(std::memory_order_relaxed);
        uint64_t next_read = current_read + 1;
        
        // 检查队列是否空
        uint64_t cached_write = cached_write_pos_.load(std::memory_order_acquire);
        if (cached_write <= current_read) {
            // 更新缓存的写位置
            cached_write = write_pos_.load(std::memory_order_acquire);
            cached_write_pos_.store(cached_write, std::memory_order_relaxed);
            
            if (cached_write <= current_read) {
                return false;  // 队列空
            }
        }
        
        // 读取数据
        item = buffer_[current_read & (Size - 1)];
        
        // 更新读位置（发布）
        read_pos_.store(next_read, std::memory_order_release);
        
        return true;
    }
};

// 无锁哈希表（使用原子操作）
#include <atomic>
#include <array>

template<typename Key, typename Value, size_t BucketCount>
class LockFreeHashMap {
private:
    struct Node {
        Key key;
        Value value;
        std::atomic<Node*> next;
        
        Node(const Key& k, const Value& v) 
            : key(k), value(v), next(nullptr) {}
    };
    
    std::array<std::atomic<Node*>, BucketCount> buckets_;
    
    size_t hash(const Key& key) const {
        return std::hash<Key>{}(key) % BucketCount;
    }
    
public:
    bool insert(const Key& key, const Value& value) {
        size_t bucket_idx = hash(key);
        Node* new_node = new Node(key, value);
        
        Node* head = buckets_[bucket_idx].load(std::memory_order_acquire);
        new_node->next.store(head, std::memory_order_relaxed);
        
        // CAS操作插入
        while (!buckets_[bucket_idx].compare_exchange_weak(
                   head, new_node,
                   std::memory_order_release,
                   std::memory_order_acquire)) {
            new_node->next.store(head, std::memory_order_relaxed);
        }
        
        return true;
    }
    
    bool find(const Key& key, Value& value) const {
        size_t bucket_idx = hash(key);
        Node* node = buckets_[bucket_idx].load(std::memory_order_acquire);
        
        while (node != nullptr) {
            if (node->key == key) {
                value = node->value;
                return true;
            }
            node = node->next.load(std::memory_order_acquire);
        }
        
        return false;
    }
};

// 使用FlatBuffers实现零拷贝序列化
#include "flatbuffers/flatbuffers.h"

// 定义FlatBuffer schema
// order.fbs:
// namespace trading;
// table Order {
//   id: uint64;
//   symbol: string;
//   price: double;
//   quantity: int32;
//   side: uint8;
// }

// 序列化（零拷贝）
flatbuffers::FlatBufferBuilder builder(1024);

auto symbol = builder.CreateString("AAPL");
auto order = trading::CreateOrder(
    builder, 
    12345,      // id
    symbol,     // symbol
    150.25,     // price
    100,        // quantity
    1           // side (buy)
);
builder.Finish(order);

// 获取序列化后的数据（无需额外拷贝）
uint8_t* buffer = builder.GetBufferPointer();
size_t size = builder.GetSize();

// 直接发送（零拷贝）
send(socket_fd, buffer, size, MSG_ZEROCOPY);

// 反序列化（零拷贝）
auto order = trading::GetOrder(buffer);
uint64_t id = order->id();
double price = order->price();
// 直接访问，无需拷贝

// 使用sendfile实现文件传输零拷贝
#include <sys/sendfile.h>

void send_file_zero_copy(int socket_fd, int file_fd, size_t file_size) {
    size_t offset = 0;
    size_t remaining = file_size;
    
    while (remaining > 0) {
        // sendfile：内核直接在内核空间传输数据，无需用户空间拷贝
        ssize_t sent = sendfile(socket_fd, file_fd, 
                                (off_t*)&offset, remaining);
        if (sent < 0) {
            perror("sendfile");
            break;
        }
        remaining -= sent;
    }
}

// Java零GC配置和实现
public class ZeroGCApplication {
    
    // JVM参数配置
    // -XX:+UseG1GC
    // -XX:MaxGCPauseMillis=10
    // -XX:+UnlockExperimentalVMOptions
    // -XX:+UseJVMCICompiler
    // -XX:ReservedCodeCacheSize=512m
    // -XX:InitialCodeCacheSize=64m
    
    // 使用堆外内存（Off-heap）
    private final ByteBuffer offHeapBuffer;
    
    public ZeroGCApplication() {
        // 分配堆外内存（不受GC管理）
        offHeapBuffer = ByteBuffer.allocateDirect(1024 * 1024 * 1024); // 1GB
    }
    
    // 对象池化，减少对象创建
    private final ThreadLocal<Order> orderPool = 
        ThreadLocal.withInitial(() -> new Order());
    
    public void processOrder(OrderData data) {
        // 重用对象，避免创建新对象
        Order order = orderPool.get();
        order.reset();
        order.setId(data.getId());
        order.setPrice(data.getPrice());
        
        // 处理订单（无GC压力）
        process(order);
    }
    
    // 使用Unsafe直接操作内存（高级用法）
    private static final Unsafe unsafe = getUnsafe();
    
    public long allocateOffHeap(long size) {
        return unsafe.allocateMemory(size);
    }
    
    public void freeOffHeap(long address) {
        unsafe.freeMemory(address);
    }
}

// 自定义内存池（减少malloc/free开销）
template<size_t BlockSize, size_t PoolSize>
class MemoryPool {
private:
    struct Block {
        char data[BlockSize];
        Block* next;
    };
    
    Block* free_list_;
    char pool_[PoolSize * BlockSize];
    std::atomic<size_t> allocated_count_{0};
    
public:
    MemoryPool() {
        // 初始化空闲链表
        free_list_ = reinterpret_cast<Block*>(pool_);
        for (size_t i = 0; i < PoolSize - 1; ++i) {
            Block* current = reinterpret_cast<Block*>(
                pool_ + i * BlockSize);
            Block* next = reinterpret_cast<Block*>(
                pool_ + (i + 1) * BlockSize);
            current->next = next;
        }
        reinterpret_cast<Block*>(
            pool_ + (PoolSize - 1) * BlockSize)->next = nullptr;
    }
    
    void* allocate() {
        if (free_list_ == nullptr) {
            return nullptr;  // 池已满
        }
        
        Block* block = free_list_;
        free_list_ = block->next;
        allocated_count_.fetch_add(1);
        
        return block->data;
    }
    
    void deallocate(void* ptr) {
        if (ptr == nullptr) {
            return;
        }
        
        Block* block = reinterpret_cast<Block*>(
            static_cast<char*>(ptr) - offsetof(Block, data));
        block->next = free_list_;
        free_list_ = block;
        allocated_count_.fetch_sub(1);
    }
    
    size_t get_allocated_count() const {
        return allocated_count_.load();
    }
};

// 事件驱动架构（减少线程切换）
#include <functional>
#include <queue>
#include <thread>

class EventDrivenEngine {
private:
    using EventHandler = std::function<void()>;
    
    LockFreeRingBuffer<EventHandler, 1024> event_queue_;
    std::atomic<bool> running_{true};
    std::thread event_loop_thread_;
    
public:
    void start() {
        event_loop_thread_ = std::thread([this]() {
            EventHandler handler;
            while (running_) {
                // 批量处理事件
                int batch_count = 0;
                while (event_queue_.try_pop(handler) && 
                       batch_count < 32) {
                    handler();  // 执行事件处理
                    batch_count++;
                }
                
                // 无事件时短暂休眠
                if (batch_count == 0) {
                    std::this_thread::sleep_for(
                        std::chrono::microseconds(10));
                }
            }
        });
    }
    
    void post_event(EventHandler handler) {
        while (!event_queue_.try_push(handler)) {
            // 队列满时等待
            std::this_thread::yield();
        }
    }
};

// 单线程事件循环（避免锁竞争）
class SingleThreadEventLoop {
private:
    std::queue<std::function<void()>> tasks_;
    int epoll_fd_;
    
public:
    void run() {
        epoll_fd_ = epoll_create1(0);
        
        while (true) {
            // 处理IO事件
            struct epoll_event events[64];
            int nfds = epoll_wait(epoll_fd_, events, 64, 0);
            
            for (int i = 0; i < nfds; ++i) {
                handle_io_event(events[i]);
            }
            
            // 处理任务队列
            while (!tasks_.empty()) {
                auto task = tasks_.front();
                tasks_.pop();
                task();
            }
            
            // 无事件时短暂休眠
            if (nfds == 0 && tasks_.empty()) {
                std::this_thread::sleep_for(
                    std::chrono::microseconds(100));
            }
        }
    }
};

// 风控前置设计
class RiskControlPreCheck {
private:
    // 基础风控（不依赖撮合结果）
    class BasicRiskControl {
    public:
        bool check_balance(uint64_t user_id, double amount) {
            // 余额检查（可并行）
            return get_balance(user_id) >= amount;
        }
        
        bool check_permission(uint64_t user_id, const std::string& symbol) {
            // 权限检查（可并行）
            return has_trading_permission(user_id, symbol);
        }
        
        bool check_daily_limit(uint64_t user_id, double amount) {
            // 日限额检查（可并行）
            return get_daily_traded(user_id) + amount <= 
                   get_daily_limit(user_id);
        }
    };
    
    // 复杂风控（依赖撮合结果，异步处理）
    class AdvancedRiskControl {
    public:
        void check_async(const Order& order, const MatchResult& result) {
            // 异步执行复杂风控检查
            std::thread([this, order, result]() {
                check_compliance(order, result);
                check_market_impact(order, result);
                check_regulatory_requirements(order, result);
            }).detach();
        }
    };
    
    BasicRiskControl basic_rc_;
    AdvancedRiskControl advanced_rc_;
    
public:
    // 前置风控（同步，快速）
    bool pre_check(const Order& order) {
        // 并行执行基础风控检查
        std::vector<std::future<bool>> futures;
        
        futures.push_back(std::async(std::launch::async,
            [this, order]() {
                return basic_rc_.check_balance(order.user_id, 
                                               order.amount);
            }));
        
        futures.push_back(std::async(std::launch::async,
            [this, order]() {
                return basic_rc_.check_permission(order.user_id, 
                                                  order.symbol);
            }));
        
        futures.push_back(std::async(std::launch::async,
            [this, order]() {
                return basic_rc_.check_daily_limit(order.user_id, 
                                                    order.amount);
            }));
        
        // 等待所有检查完成
        for (auto& future : futures) {
            if (!future.get()) {
                return false;  // 风控失败
            }
        }
        
        return true;  // 风控通过
    }
    
    // 后置风控（异步，不阻塞主流程）
    void post_check(const Order& order, const MatchResult& result) {
        advanced_rc_.check_async(order, result);
    }
};

// 二进制协议设计（替代HTTP/JSON）
struct OrderMessage {
    uint32_t magic;        // 魔数：0xDEADBEEF
    uint16_t version;     // 协议版本
    uint16_t msg_type;    // 消息类型
    uint32_t length;       // 消息长度
    uint64_t order_id;     // 订单ID
    char symbol[8];        // 交易对（固定8字节）
    double price;          // 价格
    int32_t quantity;      // 数量
    uint8_t side;          // 买卖方向
    uint64_t timestamp;    // 时间戳
    uint32_t checksum;     // 校验和
} __attribute__((packed));

// 序列化（零拷贝，直接内存布局）
void serialize_order(const Order& order, OrderMessage* msg) {
    msg->magic = 0xDEADBEEF;
    msg->version = 1;
    msg->msg_type = MSG_TYPE_ORDER;
    msg->length = sizeof(OrderMessage);
    msg->order_id = order.id;
    memcpy(msg->symbol, order.symbol.c_str(), 8);
    msg->price = order.price;
    msg->quantity = order.quantity;
    msg->side = order.side;
    msg->timestamp = get_timestamp();
    msg->checksum = calculate_checksum(msg);
}

// 反序列化（直接内存访问）
Order deserialize_order(const OrderMessage* msg) {
    Order order;
    order.id = msg->order_id;
    order.symbol = std::string(msg->symbol, 8);
    order.price = msg->price;
    order.quantity = msg->quantity;
    order.side = msg->side;
    return order;
}

// 批量确认机制
class BatchAcknowledgment {
private:
    struct PendingAck {
        uint64_t order_id;
        uint64_t timestamp;
    };
    
    std::vector<PendingAck> pending_acks_;
    std::mutex mutex_;
    std::chrono::microseconds batch_interval_{100};  // 100微秒
    std::chrono::microseconds max_wait_time_{1000};   // 1毫秒
    
public:
    void add_ack(uint64_t order_id) {
        std::lock_guard<std::mutex> lock(mutex_);
        pending_acks_.push_back({order_id, get_timestamp()});
        
        // 达到批量大小时立即发送
        if (pending_acks_.size() >= 32) {
            flush_batch();
        }
    }
    
    void flush_batch() {
        if (pending_acks_.empty()) {
            return;
        }
        
        // 批量发送确认
        send_batch_ack(pending_acks_);
        pending_acks_.clear();
    }
    
    // 定时刷新（后台线程）
    void start_batch_timer() {
        std::thread([this]() {
            while (true) {
                std::this_thread::sleep_for(batch_interval_);
                
                std::lock_guard<std::mutex> lock(mutex_);
                if (!pending_acks_.empty()) {
                    auto oldest = pending_acks_.front().timestamp;
                    auto now = get_timestamp();
                    
                    // 超过最大等待时间，立即发送
                    if (now - oldest >= max_wait_time_.count()) {
                        flush_batch();
                    }
                }
            }
        }).detach();
    }
};

// 订单簿数据结构优化
class OptimizedOrderBook {
private:
    // 使用红黑树（std::map）维护价格排序
    // 时间复杂度：插入O(log n)，查找O(log n)
    std::map<double, PriceLevel> bids_;  // 买单（价格从高到低）
    std::map<double, PriceLevel> asks_;  // 卖单（价格从低到高）
    
    // 价格索引（快速查找）
    std::unordered_map<double, std::map<double, PriceLevel>::iterator> 
        price_index_;
    
public:
    // 优化：使用迭代器避免重复查找
    void add_order(const Order& order) {
        auto& book = (order.side == SIDE_BUY) ? bids_ : asks_;
        
        // 查找或创建价格档位
        auto it = book.find(order.price);
        if (it == book.end()) {
            it = book.emplace(order.price, PriceLevel()).first;
        }
        
        // 添加到价格档位
        it->second.add_order(order);
    }
    
    // 优化：批量撮合
    MatchResult match_orders_batch(const std::vector<Order>& orders) {
        MatchResult result;
        
        // 批量处理，减少函数调用开销
        for (const auto& order : orders) {
            auto match = match_single_order(order);
            result.merge(match);
        }
        
        return result;
    }
};

// 数据库连接池
class DatabaseConnectionPool {
private:
    struct Connection {
        sql::Connection* conn;
        std::chrono::steady_clock::time_point last_used;
        bool in_use;
    };
    
    std::vector<Connection> connections_;
    std::mutex mutex_;
    std::condition_variable cv_;
    size_t pool_size_;
    
public:
    DatabaseConnectionPool(size_t size) : pool_size_(size) {
        // 预创建连接
        for (size_t i = 0; i < size; ++i) {
            connections_.push_back({
                create_connection(),
                std::chrono::steady_clock::now(),
                false
            });
        }
    }
    
    Connection* acquire() {
        std::unique_lock<std::mutex> lock(mutex_);
        
        // 等待可用连接
        cv_.wait(lock, [this]() {
            return std::any_of(connections_.begin(), 
                              connections_.end(),
                              [](const Connection& c) { 
                                  return !c.in_use; 
                              });
        });
        
        // 查找可用连接
        auto it = std::find_if(connections_.begin(), 
                              connections_.end(),
                              [](const Connection& c) { 
                                  return !c.in_use; 
                              });
        
        it->in_use = true;
        it->last_used = std::chrono::steady_clock::now();
        
        return &(*it);
    }
    
    void release(Connection* conn) {
        std::lock_guard<std::mutex> lock(mutex_);
        conn->in_use = false;
        cv_.notify_one();
    }
};