我用C++实现了一个非常标准的单消费者多生产者模式，并增加了队列中任务数量的限制。
Worker在单独的线程上运行消息队列。任务从生产者发送到Worker。如果队列中已经有max_num_tasks_，则生产者必须等待。
在VS2022中使用工具集v143为Windows x64编译。
下面的代码在Worker::Send()中的cv_.wait(lock, [&] {return (mq.Size() < max_num_tasks_); });语句处偶尔挂起。有人知道我做错了什么吗？
工人的实现

#pragma once
#include <queue>
#include <mutex>
#include <functional>

class Worker {
public:
    ~Worker() {
        Send([this] {done = true; });
        thd.join();
    }

    Worker(size_t max_num_tasks) : max_num_tasks_(max_num_tasks), done(false), thd([this] {
        while (!done) {
            mq.PopFront()();
            cv_.notify_all();
        }
        })
    { }

    void Send(std::function<void()>&& m) {
        {
            std::unique_lock<std::mutex> lock(m_);
            cv_.wait(lock, [&] {return (mq.Size() < max_num_tasks_); });
        }
        mq.PushBack(std::move(m));
    }
private:
    bool done;
    size_t max_num_tasks_;
    ThreadSafeQueue<std::function<void()>> mq;
    std::thread thd;
    std::mutex m_;
    std::condition_variable cv_;
};

线程安全队列的实现

#pragma once
#include <queue>
#include <mutex>
#include <functional>

template <typename T>
class ThreadSafeQueue {
public:
    void PushBack(T&& val) {
        {
            std::unique_lock<std::mutex> lock(q_mutex);
            q.push(std::move(val));
        }
        cv.notify_one();
    }

    void PushBack(const T& val) {
        {
            std::unique_lock<std::mutex> lock(q_mutex);
            q.push(val);
        }
        cv.notify_one();
    }

    T PopFront() {
        std::unique_lock<std::mutex> lock(q_mutex);
        cv.wait(lock, [&] { return (!q.empty()); });
        T v = q.front();
        q.pop();
        return std::move(v);
    }

    bool Empty() const {
        std::unique_lock<std::mutex> lock(q_mutex);
        return q.empty();
    }

    size_t Size() const {
        std::unique_lock<std::mutex> lock(q_mutex);
        return q.size();
    }
private:
    mutable std::mutex q_mutex;
    std::condition_variable cv;
    std::queue<T> q;
};

偶尔触发问题的单元测试示例：

TEST(WorkerTests, Compute_PI) {

        std::atomic<double> value;
        auto multiply_by_pi_over_four = [&value] {
            int n = 750;
            double v = 0;
            for (int i = 0; i < n; i++) {
                v += std::pow(-1, i) / (2 * i + 1);
            }
            value = value * v;
        };

        auto add_pi_over_four = [&value] {
            int n = 750;
            double v = 0;
            for (int i = 0; i < n; i++) {
                v += std::pow(-1, i) / (2 * i + 1);
            }
            value = value + v;
        };

        auto add_one = [&value] {
            value = value + 1;
            std::this_thread::sleep_for(std::chrono::milliseconds(3));
        };

        auto multiply_by_three = [&value] {
            value = value * 3;
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        };

        for (int i = 0; i < 100; i++) {
            value = 0.0;
            {
                Worker worker(1);
                worker.Send(add_one);
                worker.Send(multiply_by_pi_over_four);
                worker.Send(multiply_by_three);
                worker.Send([]() {});
                worker.Send(add_pi_over_four);
            }
            EXPECT_GE(3.15, value.load());
            EXPECT_LE(3.14, value.load());
        }
    }