背景
生产者和消费者问题是线程模型中的经典问题：生产者和消费者在同一时间段内共用同一个存储空间，生产者往存储空间中添加产品，消费者从存储空间中取走产品，当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。
为什么要使用生产者和消费者模式：
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题，于是引入了生产者和消费者模式。
生产者——消费者模型 会遇到的问题
生产者——消费者模型中，生产者和消费者线程之间需要传递一定量的数据，两个线程会使用一个特定大小的共享环形缓冲器。
生产者向缓冲器中写入数据，直到它到达缓冲器的终点；然后它会再次从起点重新开始，覆盖已经存在的数据。消费者线程则会读取生成的数据。
在生产者——消费者实例中，对于同步的需求有两个部分：
1.如果生产者线程生成数据的速度太快，那么将会把消费者线程还没有读取的数据覆盖；
2.如果消费者线程读取数据的速度过快，那么它就会越过生产者线程而读取一些垃圾数据。
生产者——消费者模型 解决方法：
容易导致死锁的实现：
让生产者线程填满缓冲器，然后等待消费者线程读取完缓冲器中全部数据。
使用2个信号量 解决 单消费者单生产者的问题
使用两个信号量：freeSpace 与 usedSpace
freeSpace 信号量控制生产者线程写入数据的那部分缓冲器， usedSpace 信号量则控制消费者线程读取数据的那部分缓冲器区域。
这两个区域是相互补充的。
常用缓冲区容量值初始化 freeSpace 信号量，意味着它最多可以获取的缓冲器资源量。
在启动这个应用程序时，消费者线程就会获得自由的字节并把它们转换为用过的字节。
用0初始化usedSpace信号量，以确保消费者线程不会在一开始就读取到垃圾数据。
在生产者线程中，每次反复写入都是从获取一个 freeSpace 开始。
如果该缓冲器中充满了消费者线程还没有读取的数据，那么对acquire()的调用就会被阻塞，直到消费者线程开始消费这些数据。
一旦生产者线程获取这一字节，就写入数据，并将这个字节释放为 usedSpace ，以让消费者线程读取到。
在消费者线程中，我们从获取一个 usedSpace 开始。
如果缓冲器中还没有任何可用的数据，那么将会阻塞对acquire()调用，直到生产者线程生产数据。
一旦获取到这个字节，就使用数据，并把字节释放为 freeSpace ，这样，生产者线程就可以再次写入。
上述方法在只有一个生产者和一个消费者时能解决问题。对于多个生产者或者多个消费者共享缓冲区的情况，该算法也会导致竞争条件，出现两个或以上的进程同时读或写同一个缓冲区槽的情况。
解决多生产者、多消费者 的问题
为了保证同一时刻只有一个生产者能够执行 putItemIntoBuffer()。也就是说，需要寻找一种方法来互斥地执行临界区的代码。为了达到这个目的，可引入一个二值信号灯 mutex，其值只能为 1 或者 0。如果把线程放入 down(mutex) 和 up(mutex) 之间，就可以限制只有一个线程能被执行。
#include <QtCore/QCoreApplication>
#include <QThread>
#include <QSemaphore>
#include <QDebug>
const int SIZE = 5;
static unsigned char g_buff[SIZE] = {0};
static QSemaphore g_sem_free(SIZE);
static QSemaphore g_sem_used(0);
static QSemaphore mutex(1);
class Producer : public QThread
{
protected:
void run()
{
while( true )
{
int value = qrand() % 256;
g_sem_free.acquire();
mutex.acquire();
for(int i=0; i<SIZE; i++)
{
if( !g_buff[i] )
{
g_buff[i] = (unsigned char)value;
qDebug() << objectName() << " generate: {" << i << ", " << value << "}";
break;
}
}
mutex.release();
g_sem_used.release();
sleep(2);
}
}
};
class Customer : public QThread
{
protected:
void run()
{
while( true )
{
g_sem_used.acquire();
mutex.acquire();
for(int i=0; i<SIZE; i++)
{
if( g_buff[i] )
{
int value = g_buff[i];
g_buff[i] = 0;
qDebug() << objectName() << " consume: {" << i << ", " << value << "}";
break;
}
}
mutex.release();
g_sem_free.release();
sleep(1);
}
}
};
int main(int argc, char *argv[])
{
QCoreApplication a(argc, argv);
Producer p1;
Producer p2;
Producer p3;
p1.setObjectName("p1");
p2.setObjectName("p2");
p3.setObjectName("p3");
Customer c1;
Customer c2;
c1.setObjectName("c1");
c2.setObjectName("c2");
p1.start();
p2.start();
p3.start();
c1.start();
c2.start();
return a.exec();
}
使用 互斥量 + 条件变量 解决 多生产者多消费者的问题
#include <QtCore/QCoreApplication>
#include <QWaitCondition>
#include <QThread>
#include <QMutex>
#include <iostream>
const int DataSize = 100;
const int BufferSize = 1;
// static char buffer[BufferSize];
static QWaitCondition bufferIsNotFull;
static QWaitCondition bufferIsNotEmpty;
static QMutex mutex;
static int usedSpace;
class Producer : public QThread
{
protected:
void run()
{
for (int i = 0; i < DataSize; ++i)
{
mutex.lock();
while (usedSpace == BufferSize)
{
bufferIsNotFull.wait(&mutex);
}
std::cerr<<"P";
++usedSpace;
bufferIsNotEmpty.wakeAll();
mutex.unlock();
}
}
};
class Consumer : public QThread
{
protected:
void run()
{
for (int i = 0; i < DataSize; ++i)
{
mutex.lock();
while (usedSpace == 0)
{
bufferIsNotEmpty.wait(&mutex);
}
std::cerr<<"C";
--usedSpace;
bufferIsNotFull.wakeAll();
mutex.unlock();
}
std::cerr<<std::endl;
}
};
int main(int argc, char *argv[])
{
QCoreApplication a(argc, argv);
Producer producer;
Consumer consumer;
producer.start();
consumer.start();
producer.wait();
consumer.wait();
return a.exec();
}