使用信号量的生产者和消费者

The Producer and Consumer using Semaphores example shows how to use QSemaphore to control access to a circular buffer shared by a producer thread and a consumer thread.

生产者把数据写入缓冲直到达到缓冲末端为止，此时它从头重新开始，覆盖现有数据。消费者线程读取产生数据并将其写入标准错误。

信号量使之相比互斥拥有更高级的并发成为可能。若对缓冲的访问守卫通过 QMutex ，消费者线程与生产者线程无法同时访问缓冲。然而，没有坏处让 2 线程操控 不同部分 的缓冲在同一时间。

范例包含 2 个类： Producer and Consumer 。两者继承自 QThread 。在这 2 个类之间进行通信所使用的循环缓冲和保护它的信号量都是全局变量。

替代使用 QSemaphore 以解决生产者-消费者问题是使用 QWaitCondition and QMutex . This is what the 使用等待条件的生产者和消费者 example does.

全局变量

让我们从审查循环缓冲和关联信号量开始：

constexpr int DataSize = 100000;
constexpr int BufferSize = 8192;
char buffer[BufferSize];
QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

DataSize is the amount of data that the producer will generate. To keep the example as simple as possible, we make it a constant. BufferSize is the size of the circular buffer. It is less than DataSize , meaning that at some point the producer will reach the end of the buffer and restart from the beginning.

To synchronize the producer and the consumer, we need two semaphores. The freeBytes semaphore controls the "free" area of the buffer (the area that the producer hasn't filled with data yet or that the consumer has already read). The usedBytes semaphore controls the "used" area of the buffer (the area that the producer has filled but that the consumer hasn't read yet).

Together, the semaphores ensure that the producer is never more than BufferSize bytes ahead of the consumer, and that the consumer never reads data that the producer hasn't generated yet.

The freeBytes 信号量被初始化采用 BufferSize , because initially the entire buffer is empty. The usedBytes semaphore is initialized to 0 (the default value if none is specified).

Producer (生产者) 类

让我们审查代码为 Producer 类：

class Producer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            freeBytes.acquire();
            buffer[i % BufferSize] = "ACGT"[QRandomGenerator::global()->bounded(4)];
            usedBytes.release();
        }
    }
};

生产者生成 DataSize bytes of data. Before it writes a byte to the circular buffer, it must acquire a "free" byte using the freeBytes semaphore. The QSemaphore::acquire () call might block if the consumer hasn't kept up the pace with the producer.

At the end, the producer releases a byte using the usedBytes semaphore. The "free" byte has successfully been transformed into a "used" byte, ready to be read by the consumer.

Consumer (消费者) 类

让我们现在转到 Consumer 类：

class Consumer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            usedBytes.acquire();
            fprintf(stderr, "%c", buffer[i % BufferSize]);
            freeBytes.release();
        }
        fprintf(stderr, "\n");
    }
};

The code is very similar to the producer, except that this time we acquire a "used" byte and release a "free" byte, instead of the opposite.

main() 函数

在 main() ，我们创建 2 线程并调用 QThread::wait () 以确保 2 线程在退出之前都有时间完成：

int main(int argc, char *argv[])
{
    QCoreApplication app(argc, argv);
    Producer producer;
    Consumer consumer;
    producer.start();
    consumer.start();
    producer.wait();
    consumer.wait();
    return 0;
}

So what happens when we run the program? Initially, the producer thread is the only one that can do anything; the consumer is blocked waiting for the usedBytes semaphore to be released (its initial available () count is 0). Once the producer has put one byte in the buffer, freeBytes.available() is BufferSize - 1 and usedBytes.available() is 1. At that point, two things can happen: Either the consumer thread takes over and reads that byte, or the producer thread gets to produce a second byte.

The producer-consumer model presented in this example makes it possible to write highly concurrent multithreaded applications. On a multiprocessor machine, the program is potentially up to twice as fast as the equivalent mutex-based program, since the two threads can be active at the same time on different parts of the buffer.

Be aware though that these benefits aren't always realized. Acquiring and releasing a QSemaphore has a cost. In practice, it would probably be worthwhile to divide the buffer into chunks and to operate on chunks instead of individual bytes. The buffer size is also a parameter that must be selected carefully, based on experimentation.

范例工程 @ code.qt.io