所有生产者结束后如何正确终止多消费者：消费完缓冲区剩余数据后优雅退出

原实现的核心阻塞原因 #

原代码运行到最后出现永久阻塞，本质是三个逻辑漏洞：

• 全局producerAlive标志无任何同步机制保护，既没有内存屏障，也没有被互斥锁保护，编译器可能将该值缓存到寄存器中，导致消费者线程永远无法读到生产者设置的最终状态，存在严重的内存可见性问题
• 唤醒逻辑缺失：当最后一条缓冲区中的真实数据被取出、缓冲区完全为空时，剩余仍阻塞在sem_wait(&bufferHasData)上的消费者永远等不到新的sem_post调用（生产者已经终止生产，不会再产生新数据触发信号量递增），因此会永久挂起
• 原实现拆分为mutexFill和mutexEmpty两个互斥信号量属于冗余设计，两个锁保护的临界区都是操作环形缓冲区的读写指针，完全可以合并为一个锁降低复杂度

正确的实现思路 #

核心原则是：生产者生产结束后，必须主动给所有阻塞中的消费者发送可感知的退出唤醒信号，不能指望靠现有数据信号量自然唤醒所有消费者。具体落地逻辑：

1. 用受互斥锁保护的变量记录两个核心状态：环形缓冲区的读写指针、当前缓冲区中剩余的真实待消费数据数量remainingItems
2. 所有涉及读写指针、剩余数据数的操作都必须持锁执行，避免竞态条件
3. 生产者完成全部既定数据的生产、设置好生产完成状态后，按照消费者总数量，循环执行consumerAmount次sem_post(&bufferHasData)——这部分post操作不对应真实的待消费数据，仅作为退出唤醒信号，保证每个阻塞在信号量上的消费者都能被唤醒
4. 消费者被信号量唤醒后，先持锁检查当前的remainingItems值：如果remainingItems > 0，说明是真实待消费数据，取走数据后递减计数；如果remainingItems == 0且生产已经完成，说明收到的是退出唤醒信号，直接跳出循环退出

可运行的最小修改版本代码 #

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <pthread.h>
#include <stdbool.h>
#define bufferSize 6
#define consumerAmount 3

int buffer[bufferSize];
int producerCell = 0, consumerCell = 0;
sem_t bufferHasSpace, bufferHasData;
pthread_mutex_t bufferMutex; // 统一的缓冲区互斥锁，替代原有的两个冗余信号量
int remainingItems = 0; // 缓冲区中剩余的真实待消费数据数量
bool productionFinished = false; // 受锁保护的生产完成标记

void *producer_routine()
{
    int producedCell;
    for (int i = 0; i < 20; i++)
    {
        // 等待缓冲区有空闲空间
        sem_wait(&bufferHasSpace);
        
        pthread_mutex_lock(&bufferMutex);
        producedCell = producerCell;
        producerCell = (producerCell + 1) % bufferSize;
        buffer[producedCell] = i;
        remainingItems++;
        pthread_mutex_unlock(&bufferMutex);
        
        printf("Producer generated %i\n", i);
        // 唤醒等待数据的消费者
        sem_post(&bufferHasData);
    }

    pthread_mutex_lock(&bufferMutex);
    productionFinished = true;
    pthread_mutex_unlock(&bufferMutex);
    printf("Producer ended\n");

    // 给每个消费者补发一个唤醒信号，让所有阻塞等待的消费者都能被唤醒
    for (int i = 0; i < consumerAmount; i++)
    {
        sem_post(&bufferHasData);
    }
    return NULL;
}

void *consumer_routine(void *id)
{
    int consumerID = *(int *)id;
    int consumedValue;
    while (1)
    {
        // 等待数据唤醒（可能是真实数据，也可能是生产者发的退出信号）
        sem_wait(&bufferHasData);
        
        pthread_mutex_lock(&bufferMutex);
        // 没有剩余真实数据，且生产已经完成，说明是退出唤醒
        if (remainingItems == 0 && productionFinished)
        {
            pthread_mutex_unlock(&bufferMutex);
            break;
        }
        // 取走真实数据
        int consumedCell = consumerCell;
        consumedValue = buffer[consumedCell];
        consumerCell = (consumerCell + 1) % bufferSize;
        remainingItems--;
        pthread_mutex_unlock(&bufferMutex);

        printf("Consumer %i processed %i\n", consumerID, consumedValue);
        // 归还缓冲区空间
        sem_post(&bufferHasSpace);
    }
    printf("Consumer %i ended\n", consumerID);
    return NULL;
}

int main()
{
    sem_init(&bufferHasSpace, 1, bufferSize);
    sem_init(&bufferHasData, 1, 0);
    pthread_mutex_init(&bufferMutex, NULL);
    
    pthread_t consumers[consumerAmount];
    int consumerIDs[consumerAmount];
    for (int i = 0; i < consumerAmount; i++)
    {
        consumerIDs[i] = i;
        pthread_create(&consumers[i], NULL, &consumer_routine, &consumerIDs[i]);
    }
    
    pthread_t producer;
    pthread_create(&producer, NULL, &producer_routine, NULL);
    pthread_join(producer, NULL);

    for (int i = 0; i < consumerAmount; i++)
    {
        pthread_join(consumers[i], NULL);
    }

    sem_destroy(&bufferHasSpace);
    sem_destroy(&bufferHasData);
    pthread_mutex_destroy(&bufferMutex);
    return 0;
}

之前尝试方案失败的原因 #

• 仅靠bufferHasData信号量判断退出：信号量的等待队列是FIFO的，一旦信号量计数降到0，所有仍在队列中阻塞的消费者永远不会被唤醒，无法匹配退出时机
• 原代码的布尔标志+信号量组合：最后一条真实数据被取走后，没有任何触发信号量post的动作，其余阻塞的消费者永远等不到唤醒，会一直挂起
• 仅靠布尔标志判断退出：消费者没有等待缓冲区的剩余数据被全部消费就直接退出，会丢失未处理的生产数据
• 无终止逻辑的无限循环：没有设计明确的退出出口，消费者永远不会主动结束运行

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