如何在C++20中用std::counting_semaphore实现多单元获取/释放的生产者消费者场景

根据Dijkstra最初定义的信号量，支持一次性等待或释放多个资源单元，其核心操作如下：

function V(semaphore S, integer I):
    [S ← S + I]

function P(semaphore S, integer I):
    repeat:
        [if S ≥ I:
        S ← S − I
        break]

但C++20引入的std::counting_semaphore仅支持单次操作1个资源单元（acquire()无参数），这给需要批量操作的场景（比如消费者需等待缓冲区有指定数量物品才执行）带来不便。

当前的解决方案是循环调用acquire()/release()，但这种方式繁琐且不够高效：

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>
#include <semaphore>
#include <cstdlib>

#define MAX 100
static std::vector<int> buffer;
static std::mutex mtx;
static std::counting_semaphore<MAX> sem_empty(MAX);
static std::counting_semaphore<MAX> sem_full(0);

void consumer(int N) {
    while (true) {
        for (int i = 0; i < N; ++i) {
            sem_full.acquire();
        }
        {
            std::lock_guard<std::mutex> lock(mtx);
            for (int i = 0; i < N; ++i) {
                if (!buffer.empty()) {
                    int data = buffer.back();
                    buffer.pop_back();
                    std::cout << "Consumer: " << data << std::endl;
                }
            }
        }
        for (int i = 0; i < N; ++i) {
            sem_empty.release();
        }
    }
}

void producer() {
    while (true) {
        const int data = std::rand() % 100;
        sem_empty.acquire();
        {
            std::lock_guard<std::mutex> lock(mtx);
            buffer.push_back(data);
            std::cout << "Producer: " << data << std::endl;
        }
        sem_full.release();
    }
}

int main() {
    std::thread t1(producer);
    std::thread t2(consumer, 4); // 消费者每次消费4个物品
    t1.join();
    t2.join();
}

请问是否有更优雅高效的方式，用C++20标准库实现批量资源操作？或是有更合适的同步机制？

1. 优化现有信号量代码：利用release的批量参数 #

首先纠正一个常见误区：std::counting_semaphore::release支持传入整数参数，可以一次性释放多个资源单元，不需要循环调用。仅acquire()需要循环实现批量等待，我们可以把这部分逻辑封装成辅助函数，让代码更简洁：

#include <semaphore>

// 封装批量等待逻辑
template<std::ptrdiff_t MaxCount>
void acquire_batch(std::counting_semaphore<MaxCount>& sem, std::ptrdiff_t count) {
    for (std::ptrdiff_t i = 0; i < count; ++i) {
        sem.acquire();
    }
}

修改后的消费者代码会更简洁高效：

void consumer(int N) {
    while (true) {
        acquire_batch(sem_full, N);
        
        {
            std::lock_guard<std::mutex> lock(mtx);
            for (int i = 0; i < N; ++i) {
                if (!buffer.empty()) {
                    int data = buffer.back();
                    buffer.pop_back();
                    std::cout << "Consumer: " << data << std::endl;
                }
            }
        }
        
        sem_empty.release(N); // 直接批量释放，无需循环
    }
}

2. 换用std::condition_variable（更适配批量等待场景） #

对于需要等待“缓冲区达到指定数量”的生产者消费者场景，std::condition_variable配合std::mutex是更直观高效的选择，它可以直接等待目标条件，避免信号量循环调用的开销：

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>
#include <condition_variable>
#include <cstdlib>

#define MAX 100
static std::vector<int> buffer;
static std::mutex mtx;
static std::condition_variable cv_not_empty;
static std::condition_variable cv_not_full;

void consumer(int N) {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        // 直接等待缓冲区至少有N个元素
        cv_not_empty.wait(lock, [N](){ return buffer.size() >= N; });
        
        // 批量消费
        for (int i = 0; i < N; ++i) {
            int data = buffer.back();
            buffer.pop_back();
            std::cout << "Consumer: " << data << std::endl;
        }
        
        // 通知生产者缓冲区有空闲空间
        cv_not_full.notify_all();
    }
}

void producer() {
    while (true) {
        const int data = std::rand() % 100;
        
        std::unique_lock<std::mutex> lock(mtx);
        cv_not_full.wait(lock, [](){ return buffer.size() < MAX; });
        
        buffer.push_back(data);
        std::cout << "Producer: " << data << std::endl;
        
        // 通知消费者缓冲区有元素
        cv_not_empty.notify_all();
    }
}

int main() {
    std::thread t1(producer);
    std::thread t2(consumer, 4);
    t1.join();
    t2.join();
}

这种方式语义清晰，直接对应“等待缓冲区满足数量条件”的业务逻辑，无需拆分资源操作。

3. 自定义批量信号量封装（贴合Dijkstra语义） #

如果坚持使用信号量风格的API，可以基于std::counting_semaphore封装一个完全支持批量P/V操作的类：

#include <semaphore>

template<std::ptrdiff_t MaxCount>
class BatchCountingSemaphore {
public:
    explicit BatchCountingSemaphore(std::ptrdiff_t initial_count) : sem_(initial_count) {}
    
    // 对应Dijkstra的P操作：批量等待
    void P(std::ptrdiff_t count) {
        for (std::ptrdiff_t i = 0; i < count; ++i) {
            sem_.acquire();
        }
    }
    
    // 对应Dijkstra的V操作：批量释放
    void V(std::ptrdiff_t count) {
        sem_.release(count);
    }
    
private:
    std::counting_semaphore<MaxCount> sem_;
};

使用时完全贴合原始信号量语义：

static BatchCountingSemaphore<MAX> sem_empty(MAX);
static BatchCountingSemaphore<MAX> sem_full(0);

void consumer(int N) {
    while (true) {
        sem_full.P(N);
        
        {
            std::lock_guard<std::mutex> lock(mtx);
            // 批量消费逻辑...
        }
        
        sem_empty.V(N);
    }
}

内容的提问来源于stack exchange，提问作者nowox