基于C++ STL的生产者-多消费者模型：生产者需等空闲消费者再生产

嘿，这个需求我太懂了！轮询不仅浪费CPU资源，还容易让队列无限制堆积，咱们用同步原语就能完美实现你要的效果——生产者只在有消费者空闲时才生产数据，同一时刻最多存在3个未处理的数据。

核心思路 #

核心是用计数信号量来控制生产者的生产节奏，搭配条件变量实现消费者的等待通知，全程不需要轮询：

• 初始化一个计数信号量，值等于消费者数量（这里是3），它代表当前可用的"消费槽位"
• 生产者每次生产前必须获取这个信号量：如果所有消费者都在忙（槽位用完），生产者会自动阻塞，直到有消费者释放槽位
• 消费者处理完数据后释放信号量，通知生产者可以继续生产
• 用条件变量让消费者在没有数据时阻塞等待，避免轮询检查队列

Python 代码示例 #

下面是一个可运行的实现，用threading模块的同步原语：

import threading
import time
import random

# 信号量：初始值3，对应3个消费者的空闲槽位
slot_semaphore = threading.Semaphore(3)
# 条件变量：用于通知消费者有新数据可用
data_condition = threading.Condition()
# 共享数据队列（最多容纳3个未处理数据）
data_queue = []

def producer():
    while True:
        # 先获取空闲槽位，无空闲则阻塞，无需轮询
        slot_semaphore.acquire()
        
        # 生成数据
        new_data = random.randint(1, 100)
        print(f"✅ 生产者生成数据: {new_data}")
        
        # 加锁更新队列并通知消费者
        with data_condition:
            data_queue.append(new_data)
            data_condition.notify()
        
        # 模拟生产耗时
        time.sleep(random.uniform(0.3, 1.2))

def consumer(consumer_id):
    while True:
        with data_condition:
            # 等待直到队列中有数据，避免轮询
            while not data_queue:
                data_condition.wait()
            # 取出数据
            target_data = data_queue.pop(0)
        
        print(f"👥 消费者{consumer_id}处理数据: {target_data}")
        
        # 处理完成后释放槽位，让生产者可以继续生产
        slot_semaphore.release()
        
        # 模拟消费耗时
        time.sleep(random.uniform(0.8, 1.8))

# 启动3个消费者线程
for idx in range(3):
    threading.Thread(target=consumer, args=(idx+1,), daemon=True).start()

# 启动生产者线程
threading.Thread(target=producer, daemon=True).start()

# 主线程保持运行
while True:
    time.sleep(1)

关键细节解释 #

• slot_semaphore：完美限制了未处理数据的最大数量（3个），生产者只有在有消费者空闲时才能获取信号量生产新数据，完全符合你的需求
• 条件变量data_condition：让消费者在没有数据时进入阻塞状态，有新数据时被唤醒，彻底避免轮询检查队列的操作
• 全程都是基于操作系统级别的同步机制，比轮询高效得多，也更优雅

其他语言的实现思路 #

不管是Java、C++还是Go，核心逻辑都是一样的：

• Java 用java.util.concurrent.Semaphore配合Condition
• C++ 用std::counting_semaphore配合std::condition_variable
• Go 用带缓冲的通道（缓冲大小设为3），通道本身就自带了类似信号量的阻塞机制，实现起来更简洁

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