条件变量队列

Condition Variable Queue

专题: Systems & Architecture / 系统与架构
难度: L1
来源: MyntBit

个人笔记

题目详情

在高频交易系统中，行情数据事件异步到达，必须由交易策略顺序处理。处理此并发的一种可靠模式是使用线程安全事件队列的生产者-消费者模型。此模式确保网络 I/O 线程和策略执行线程之间的安全数据传递，不会产生竞争条件。

任务：实现 EventQueue 类，使用 mutex 和 condition_variable 实现线程安全队列。生产者线程调用 push() 添加事件，消费者线程调用 pop() 等待并取出事件。支持超时等待和优雅关闭。

英文原题

In high-frequency trading systems, market data events often arrive asynchronously and must be processed sequentially by a trading strategy. A robust pattern for handling this concurrency is the producer-consumer model utilizing a thread-safe event queue. Implementing this pattern ensures safe data handoffs between network I/O threads and strategy execution threads without race conditions.
Task
Implement a thread-safe, blocking TradeQueue class using std::mutex and std::condition_variable.
The

解析

问题分析

条件变量允许线程在条件不满足时阻塞等待，避免忙循环消耗 CPU。在阻塞队列中，消费者在队列空时等待条件变量，生产者入队后通知消费者唤醒。

实现

template<typename T>
class BlockingQueue {
    std::queue<T> q_; std::mutex mtx_; std::condition_variable cv_;
    size_t max_size_; bool closed_{false};
public:
    void push(T v) {
        std::unique_lock lk(mtx_);
        cv_.wait(lk, [this]{ return q_.size() < max_size_ || closed_; });
        if (!closed_) { q_.push(std::move(v)); cv_.notify_one(); }
    }
    std::optional<T> pop() {
        std::unique_lock lk(mtx_);
        cv_.wait(lk, [this]{ return !q_.empty() || closed_; });
        if (q_.empty()) return std::nullopt;
        T v = std::move(q_.front()); q_.pop();
        cv_.notify_one();  // 通知等待空间的生产者
        return v;
    }
    void close() { std::lock_guard lk(mtx_); closed_ = true; cv_.notify_all(); }
};

复杂度与边界

时间复杂度：push/pop O(1) 均摊
空间复杂度：O(队列容量)
边界条件：(1) 虚假唤醒需用 predicate lambda (2) 析构前必须 close + join (3) 队列满时生产者阻塞实现背压

英文解析

Analysis

Condition variables allow threads to block when a condition is not satisfied, avoiding busy-wait CPU consumption. In a blocking queue, consumers wait on the condition variable when the queue is empty, and producers notify consumers after enqueuing an item.

Solution

template<typename T>
class BlockingQueue {
    std::queue<T> q_; std::mutex mtx_; std::condition_variable cv_;
    size_t max_size_; bool closed_{false};
public:
    void push(T v) {
        std::unique_lock lk(mtx_);
        cv_.wait(lk, [this]{ return q_.size() < max_size_ || closed_; });
        if (!closed_) { q_.push(std::move(v)); cv_.notify_one(); }
    }
    std::optional<T> pop() {
        std::unique_lock lk(mtx_);
        cv_.wait(lk, [this]{ return !q_.empty() || closed_; });
        if (q_.empty()) return std::nullopt;
        T v = std::move(q_.front()); q_.pop();
        cv_.notify_one();
        return v;
    }
    void close() { std::lock_guard lk(mtx_); closed_ = true; cv_.notify_all(); }
};

Complexity & Edge Cases

Time complexity: push/pop O(1) amortized
Space complexity: O(queue capacity)
Edge cases: (1) Spurious wakeups require predicate lambda (2) Must close + join before destruction (3) Full-queue producer blocking provides backpressure

Verification

Test concurrent push/pop with multiple producer and consumer threads. Verify no data loss, proper blocking when queue is full/empty, and clean shutdown via close().

Key Considerations

The predicate-based wait pattern eliminates spurious wakeup issues. In trading systems, bounded blocking queues provide natural backpressure - when the order gateway cannot process fast enough, strategy threads block on push, preventing unbounded order accumulation.