12.4 ZMQ消息中间件

基于订单簿设计中的实现与ZeroMQ的简单消息中心

#include <atomic>
#include <chrono>
#include <iostream>
#include <mutex>
#include <sstream>
#include <string>
#include <thread>
#include <zmq.hpp>

#include "circular_array_orderbook.hpp"

#ifdef __linux__
#include <pthread.h>
#include <sched.h>
#endif

namespace json {
// 序列化单个订单
std::string serialize_order(const circular_array::Order& order) {
  std::stringstream ss;
  ss << "{" << "\"id\":" << order.id << "," << "\"price\":" << order.price
     << "," << "\"quantity\":" << order.quantity << "}";
  return ss.str();
}

// 序列化单个价格档位
template <typename OrderContainer>
std::string serialize_price_level(
    const circular_array::PriceLevel<OrderContainer>& level) {
  std::stringstream ss;
  ss << "{" << "\"price\":" << level.price << ","
     << "\"total_quantity\":" << level.total_quantity << ","
     << "\"orders\":[";  // 嵌套订单列表

  // 遍历档位下所有订单
  auto it = level.orders.begin();
  while (it != level.orders.end()) {
    ss << serialize_order(*it);
    ++it;
    if (it != level.orders.end()) {
      ss << ",";  // 最后一个订单后不加逗号
    }
  }

  ss << "]}";
  return ss.str();
}

// 序列化单个方向的所有档位
template <typename Container>
std::string serialize_side(const Container& side, int depth) {
  std::stringstream ss;
  ss << "[";

  bool has_valid_level = false;
  for (int i = 0; i < depth; ++i) {
    const auto& level = side[i];
    if (level.total_quantity > 0) {
      if (has_valid_level) {
        ss << ",";
      }
      ss << serialize_price_level(level);
      has_valid_level = true;
    }
  }

  ss << "]";
  return ss.str();
}

// 序列化完整订单簿快照（买盘+卖盘+元数据）
template <typename OrderBookType>
std::string serialize_full_book(const OrderBookType& order_book) {
  // 获取订单簿核心参数
  int depth = order_book.get_depth();
  int precision = order_book.get_precision();
  auto timestamp = std::chrono::duration_cast<std::chrono::microseconds>(
                       std::chrono::system_clock::now().time_since_epoch())
                       .count();

  std::stringstream ss;
  ss << "{" << "\"timestamp\":" << timestamp << ","
     << "\"precision\":" << precision << "," << "\"depth\":" << depth << ","
     << "\"bids\":" << serialize_side(order_book.get_bids(), depth) << ","
     << "\"offers\":" << serialize_side(order_book.get_offers(), depth) << "}";

  return ss.str();
}
}  // namespace json

// 消息中心
template <typename OrderBookType>
class MessagingHub {
 private:
  zmq::context_t context;
  zmq::socket_t publisher;
  OrderBookType& order_book;
  std::atomic<bool> running;
  int cpu_core;
  const int snapshot_interval_us;  // 全快照发布间隔（微秒，默认100ms）

  // 状态跟踪
  struct BookState {
    circular_array::LevelData best_bid;
    circular_array::LevelData best_offer;

    bool operator!=(const BookState& other) const {
      return best_bid.price != other.best_bid.price ||
             best_bid.total_quantity != other.best_bid.total_quantity ||
             best_offer.price != other.best_offer.price ||
             best_offer.total_quantity != other.best_offer.total_quantity;
    }
  };

  mutable std::mutex state_mutex;
  BookState last_state;
  std::chrono::microseconds last_snapshot_time;  // 上一次全快照发布时间

 public:
  MessagingHub(OrderBookType& lob,
               const std::string& bind_addr = "tcp://*:5556", int core = 0,
               int snapshot_interval_us = 100000)
      : context(1),
        publisher(context, ZMQ_PUB),
        order_book(lob),
        running(false),
        cpu_core(core),
        snapshot_interval_us(snapshot_interval_us),
        last_snapshot_time(std::chrono::microseconds(0)) {
    publisher.bind(bind_addr);
    std::cout << "MessagingHub: Bound to " << bind_addr << std::endl;
    std::cout << "MessagingHub: Full snapshot interval = "
              << snapshot_interval_us / 1000 << "ms" << std::endl;

    last_state = capture_current_state();
  }

  ~MessagingHub() { stop(); }

  void start() {
    running = true;
    std::thread(&MessagingHub::run_loop, this).detach();
  }

  void stop() { running = false; }

 private:
  void run_loop() {
    if (!set_cpu_affinity(cpu_core)) {
      std::cerr << "MessagingHub: Warn - Failed to bind CPU core " << cpu_core
                << std::endl;
    } else {
      std::cout << "MessagingHub: Running on CPU core " << cpu_core
                << std::endl;
    }

    while (running) {
      const auto now = std::chrono::high_resolution_clock::now();
      const auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(
          now.time_since_epoch());

      // 1. 处理增量更新（仅关键行情变化时发布）
      handle_incremental_update();

      // 2. 处理定时全快照（达到间隔则发布）
      if (now_us - last_snapshot_time >=
          std::chrono::microseconds(snapshot_interval_us)) {
        handle_full_snapshot();
        last_snapshot_time = now_us;
      }

      std::this_thread::sleep_for(std::chrono::microseconds(100));
    }

    std::cout << "MessagingHub: Stopped" << std::endl;
  }

  // 捕获当前关键状态
  BookState capture_current_state() const {
    BookState state;
    state.best_bid = order_book.get_best_bid();
    state.best_offer = order_book.get_best_offer();
    return state;
  }

  // 处理增量更新（仅发布关键行情变化）
  void handle_incremental_update() {
    BookState current_state = capture_current_state();
    {
      std::lock_guard<std::mutex> lock(state_mutex);
      if (current_state != last_state) {
        send_zmq_message("inc_best_bid",
                         json::serialize_level(current_state.best_bid));
        send_zmq_message("inc_best_offer",
                         json::serialize_level(current_state.best_offer));
        last_state = current_state;
      }
    }
  }

  // 处理全快照发布（包含所有非空档位+订单明细）
  void handle_full_snapshot() {
    try {
      std::string full_snapshot = json::serialize_full_book(order_book);
      send_zmq_message("full_book", full_snapshot);
      std::cout << "MessagingHub: Published full snapshot (size: "
                << full_snapshot.size() << " bytes)" << std::endl;
    } catch (const std::exception& e) {
      std::cerr << "MessagingHub: Error publishing full snapshot - " << e.what()
                << std::endl;
    }
  }

  // 发送ZMQ消息（主题+内容，兼容ZMQ订阅过滤）
  void send_zmq_message(const std::string& topic, const std::string& content) {
    try {
      std::string msg = topic + " " + content;
      zmq::message_t zmq_msg(msg.size());
      memcpy(zmq_msg.data(), msg.data(), msg.size());
      publisher.send(zmq_msg, zmq::send_flags::none);
    } catch (const zmq::error_t& e) {
      std::cerr << "MessagingHub: ZMQ send error (" << topic << ") - "
                << e.what() << std::endl;
    }
  }

#ifdef __linux__
  bool set_cpu_affinity(int core_id) {
    int max_core = sysconf(_SC_NPROCESSORS_CONF) - 1;
    if (core_id < 0 || core_id > max_core) {
      std::cerr << "MessagingHub: Invalid CPU core " << core_id
                << " (max: " << max_core << ")" << std::endl;
      return false;
    }

    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(core_id, &cpuset);
    int ret =
        pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
    return ret == 0;
  }
#else
  bool set_cpu_affinity(int) { return false; }
#endif
};

消息中间件负责大型交易系统架构、集群式分布式交易系统的系统解耦和交互，分离关键路径和冷路径，以下是一个简单示例，将PMS、RMS划分为架构的上游，将OMS、EMS划分为架构的下游，其中下游往往是性能敏感的关键路径。