使用信号量的生产者-消费者模式实现

Question

家伙！我知道可以使用信号量同步原语来实现生产者－消费者模式。但是要实现多少信号量以及发生什么事件呢？

Answer 1

从概念上讲，生产者-消费者模式包括三个部分：

1. 一个或多个生产者
2. 一个或多个消费者
3. 生产对象的共享队列

生产者生成对象并将其放入队列。使用者等待直到队列中有要消费的对象，然后对每个对象执行一些操作。通常，可以将队列视为无界的（即，可以存储无限个对象）。

在最基本的层次上，需要单个mutually-exclusive (mutex)信号量作为队列的信号。此互斥锁一次仅允许一个客户端获取该互斥锁。生产者创建一个对象，获取互斥量，使该对象排队，然后释放该互斥量。使用者获取互斥锁，从队列中删除一个元素，然后释放互斥锁。

生产者的伪代码是：

produce(queue, mutex):
    object := createObject()
    mutex.acquire()
    queue.enqueue(object)
    mutex.release()

consume(queue, mutex):
    mutex.acquire()
    object := queue.pop()
    mutex.release()
    handle(object)

当我们引入有界队列的概念时（即，队列中可存储的对象数量受到限制），此概念模型变得更加复杂。这要求我们在队列中没有对象的情况下保护使用者免于从队列中弹出，并在队列已满时保护生产者免于排队。

要完成此操作，我们必须引入计数信号量的概念。然后，我们创建两个新的信号灯：

1. empty：用队列的最大容量初始化
2. full：初始化为0

当新项目入队时，full信号量增加，empty信号量减少。 full信号量跟踪队列中有多少个已填充的插槽，empty信号量跟踪队列中的空插槽的数量。当生产者加入新对象时，它会减少empty计数并增加full计数。从队列中弹出时，使用者执行相反的操作。

对于有限的用例，伪代码变为：

produce(queue, mutex, full, empty):

    object := createObject()

    empty.decrement()
    mutex.acquire()

    queue.enqueue(object)

    mutex.release()
    full.increment()

consume(queue, mutex, full, empty):

    full.decrement()
    mutex.acquire()

    object := queue.pop()

    mutex.release()
    empty.increment()

    handle(object)