条件变量与信号量搭配顺序及多消费者生产者同步实现方法

调用顺序 #

• 信号量作为消费准入门槛时，优先操作信号量，再处理需要互斥保护的条件变量逻辑
• 生产端：完成资源准备后，先操作信号量给消费端放开准入权限，再持有互斥锁等待条件变量的完成通知
• 消费端：先申请信号量获得准入权限，完成资源消费后，再持有互斥锁修改共享计数、触发条件变量通知

触发场景 #

• 信号量触发：仅用于控制消费准入，资源未就绪时信号量值为0，所有消费者调用sem_wait直接阻塞；资源就绪后生产端调用对应次数的sem_post，逐个唤醒等待的消费者
• 条件变量触发：仅用于通知生产端所有消费已完成，每次消费者完成消费后修改计数，当计数达到总消费者数时，调用条件变量通知接口唤醒生产端

首先纠正一个认知误区：POSIX信号量是计数型信号量，不是只能取0/1的二值信号量，你只要初始化时把信号量的初始值设为0，每次生产完调用3次sem_post后信号量值会变成3，刚好允许3个消费者各执行一次sem_wait通过，完全可以实现一次性放开所有消费者读取权限的需求，你原本的信号量用法思路是正确的。

另外你提到不想在消费者逻辑加互斥锁，这个需要调整：共享计数器counter需要被多个消费者修改，自增操作本身不是原子操作，多线程同时修改会出现计数丢失问题，不过互斥锁只会在修改counter的极短时间持有，读消息的逻辑完全不需要加锁，不会影响消费的并发效率。

全局变量定义 #

#define CONSUMER_CNT 3 // 总消费者数量
int message; // 生产者生成的消息
int counter = 0; // 已完成消费的消费者计数
pthread_mutex_t guard_cv = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond_all_read = PTHREAD_COND_INITIALIZER;
sem_t sem_ready_to_read; // 初始化时设置初始值为0：sem_init(&sem_ready_to_read, 0, 0);

修正后的生产者代码 #

void *Producer(void *unused)
{   
    while (1)
    {
        /* 生产消息 */
        message = rand () % 100;
        printf("生产者生成消息：%d\n", message);
                
        // 给信号量加3，放开3个消费者的准入权限
        for(int i=0; i<CONSUMER_CNT; i++) {
            sem_post(&sem_ready_to_read);
        }
        
        // 持有互斥锁等待所有消费者消费完成
        pthread_mutex_lock(&guard_cv);
        while (counter != CONSUMER_CNT)
        {
            pthread_cond_wait(&cond_all_read, &guard_cv);
        }
        // 本轮消费完成，重置计数器供下一轮使用
        counter = 0;
        pthread_mutex_unlock(&guard_cv);
    }
    return NULL;
}

消费者代码 #

void *Consumer(void *arg)
{
    int id = *(int*)arg; // 传入的消费者编号
    while(1) {
        // 申请信号量，消息未就绪时会阻塞在这里
        sem_wait(&sem_ready_to_read);

        // 读取消费消息，仅读操作不需要加锁
        printf("消费者%d读取到消息：%d\n", id, message);

        // 修改共享计数器必须加锁保护
        pthread_mutex_lock(&guard_cv);
        counter++;
        // 所有消费者都读完后通知生产者
        if(counter == CONSUMER_CNT) {
            pthread_cond_signal(&cond_all_read);
        }
        pthread_mutex_unlock(&guard_cv);
    }
    return NULL;
}

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