Qt 线程同步（QMutex、QWaitCondition、QSemaphore）

定义部分参考 http://blog.chinaunix.net/uid-21411227-id-1826740.html

“线程同步”最典型的例子就是大学时期学过的“生产者-消费者模型”（Producer-consumer model），也称有限缓冲问题（Bounded-buffer problem）。

该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

要解决该问题，就必须让生产者在缓冲区满时休眠（要么干脆就放弃数据），等到下次消费者消耗缓冲区中的数据的时候，生产者才能被唤醒，开始往缓冲区添加数据。同样，也可以让消费者在缓冲区空时进入休眠，等到生产者往缓冲区添加数据之后，再唤醒消费者。通常采用进程间通信的方法解决该问题，常用的方法有信号灯法等。如果解决方法不够完善，则容易出现死锁的情况。出现死锁时，两个线程都会陷入休眠，等待对方唤醒自己。该问题也能被推广到多个生产者和消费者的情形。

PV原语

PV操作是由P操作原语和V操作原语组成（原语是不可中断的过程），对信号量进行操作，具体定义如下：

P（S）：①将信号量S的值减1，即S=S-1；
②如果S³0，则该进程继续执行；否则该进程置为等待状态，排入等待队列。
V（S）：①将信号量S的值加1，即S=S+1；
②如果S>0，则该进程继续执行；否则释放队列中第一个等待信号量的进程。

P操作相当于申请资源，而V操作相当于释放资源。
P操作-----申请资源
V操作-----释放资源

伪代码

在理解了PV操作的的含义后，就必须讲解利用PV操作可以实现进程的两种情况：互斥和同步。根据互斥和同步不同的特点，就有利用PV操作实现互斥与同步相对固定的结构模式。这里就不详细讲解了。但生产者-消费者问题是一个有代表性的进程同步问题并不容易。但是如果我们将问题细分成三种情况进行讲解，理解难度将大大降低。

1）一个生产者，一个消费者，公用一个缓冲区。

可以作以下比喻：将一个生产者比喻为一个生产厂家，如伊利牛奶厂家，而一个消费者，比喻是学生小明，而一个缓冲区则比喻成一间好又多。第一种情况，可以理解成伊利牛奶生产厂家生产一盒牛奶，把它放在好又多一分店进行销售，而小明则可以从那里买到这盒牛奶。只有当厂家把牛奶放在商店里面后，小明才可以从商店里买到牛奶。所以很明显这是最简单的同步问题。

解题如下：

定义两个同步信号量：
empty——表示缓冲区是否为空，初值为1。
full——表示缓冲区中是否为满，初值为0。
生产者进程
while(TRUE){
生产一个产品;
P(empty);
产品送往Buffer;
V(full);
}
消费者进程
while(TRUE){
P(full);
从Buffer取出一个产品;
V(empty);
消费该产品;

2）一个生产者，一个消费者，公用n个环形缓冲区。

第二种情况可以理解为伊利牛奶生产厂家可以生产好多牛奶，并将它们放在多个好又多分店进行销售，而小明可以从任一间好又多分店中购买到牛奶。同样，只有当厂家把牛奶放在某一分店里，小明才可以从这间分店中买到牛奶。不同于第一种情况的是，第二种情况有N个分店（即N个缓冲区形成一个环形缓冲区），所以要利用指针，要求厂家必须按一定的顺序将商品依次放到每一个分店中。缓冲区的指向则通过模运算得到。

解题如下：

定义两个同步信号量：
empty——表示缓冲区是否为空，初值为n。
full——表示缓冲区中是否为满，初值为0。
设缓冲区的编号为1～n-1，定义两个指针in和out，分别是生产者进程和消费者进程使用的指针，指向下一个可用的缓冲区。

生产者进程
while(TRUE){
生产一个产品;
P(empty);
产品送往buffer（in）；
in=(in+1)mod n；
V(full);
}
消费者进程
while(TRUE){
P(full);
从buffer（out）中取出产品；
out=(out+1)mod n；
V(empty);
消费该产品;
}

3）一组生产者，一组消费者，公用n个环形缓冲区

第三种情况，可以理解成有多间牛奶生产厂家，如蒙牛，达能，光明等，消费者也不只小明一人，有许许多多消费者。不同的牛奶生产厂家生产的商品可以放在不同的好又多分店中销售，而不同的消费者可以去不同的分店中购买。当某一分店已放满某个厂家的商品时，下一个厂家只能把商品放在下一间分店。所以在这种情况中，生产者与消费者存在同步关系，而且各个生产者之间、各个消费者之间存在互斥关系,他们必须互斥地访问缓冲区。

解题如下：

定义四个信号量：
empty——表示缓冲区是否为空，初值为n。
full——表示缓冲区中是否为满，初值为0。
mutex1——生产者之间的互斥信号量，初值为1。
mutex2——消费者之间的互斥信号量，初值为1。
设缓冲区的编号为1～n-1，定义两个指针in和out，分别是生产者进程和消费者进程使用的指针，指向下一个可用的缓冲区。

生产者进程
while(TRUE){
生产一个产品;
P(empty);
P(mutex1)；
产品送往buffer（in）；
in=(in+1)mod n；
V(mutex1);
V(full);
}
消费者进程
while(TRUE){
P(full);
P(mutex2)；
从buffer（out）中取出产品；
out=(out+1)mod n；
V（mutex2）；
V(empty);

Qt实现线程同步主要有3种方式

（1）QMutex

mutex适用于简单的线程同步，即各线程按照一定的条件顺序执行，且一次仅有一个线程活跃，其余线程休眠。

（2）QWaitCondition

waitCondition适用于需要根据条件判断线程的休眠与否的同步方式。一次可以有选择的让多个线程”同时执行”、”同时休眠”或”随机挑一个线程执行或休眠”，相较mutex方式来说灵活很多。

（3）QSemaphore

semaphore适用于线程要根据一个可增减的变量值来决定是否执行或休眠的情况。其代表例子就是”生产者-消费者模型”。

Qt代码实战

（1）MutexOnly.h

#ifndef MUTEXWAITCONDITION
#define MUTEXWAITCONDITION

#include <QCoreApplication>
#include <iostream>
#include <QThread>
#include <QMutex>
#include <QWaitCondition>

using namespace std;

class MutexWaitCondition {
public:
//预计生产(或消费)数量
int loopCount;
//当前产品数量
int product;
//仓库能容纳的最大产品数量
int capacity;

QMutex mutex;
/*
The QWaitCondition class provides a condition variable for synchronizing threads
QWaitCondition类为线程同步提供了一个条件变量
*/
QWaitCondition productIsNotFull;
QWaitCondition productIsNotEmpty;

//生产者
class Producer : public QThread {
public:
Producer(MutexWaitCondition *manager) : QThread() {
this->manager = manager;
}
protected:
void run() {
for(int i=0; i<manager->loopCount; i++) {
manager->mutex.lock();
while(manager->product == manager->capacity) {
/*
bool QWaitCondition::wait(QReadWriteLock * lockedReadWriteLock, unsigned long time = ULONG_MAX)
Releases the lockedReadWriteLock and waits on the wait condition
释放该锁，并且阻塞当前线程，直到条件满足（即调用wake方法被唤醒）
*/
manager->productIsNotFull.wait(&manager->mutex);
}
manager->product++;
//cout<<"P";
cout<<i<<".P="<<manager->product<<", ";
manager->productIsNotEmpty.wakeAll();
manager->mutex.unlock();
}
}
private:
MutexWaitCondition *manager;
};

//消费者
class Consumer : public QThread {
public:
Consumer(MutexWaitCondition *manager) : QThread() {
this->manager = manager;
}
protected:
void run() {
for(int i=0; i<manager->loopCount; i++) {
manager->mutex.lock();
while(manager->product == 0) {
manager->productIsNotEmpty.wait(&manager->mutex);
}
manager->product--;
//cout<<"C";
cout<<i<<".C="<<manager->product<<", ";
manager->productIsNotFull.wakeAll();
manager->mutex.unlock();
}
}
private:
MutexWaitCondition *manager;
};

//无修饰的方法，默认是private的
public:
void test(int loopCount, int capacity)
{
this->loopCount = loopCount;
this->capacity = capacity;
this->product = 0;

Producer producer(this);
Consumer consumer(this);
//thread.start会调用thread内部的run方法
producer.start();
consumer.start();
/*
Blocks the thread until either of these conditions is met:
阻塞该线程直到所有条件都满足
*/
producer.wait();
consumer.wait();
cout<<endl;
}
};
#endif // MUTEXWAITCONDITION

（2）MutexWaitCondition.h

#ifndef MUTEXWAITCONDITION
#define MUTEXWAITCONDITION

#include <QCoreApplication>
#include <iostream>
#include <QThread>
#include <QMutex>
#include <QWaitCondition>

using namespace std;

class MutexWaitCondition {
public:
//预计生产(或消费)数量
int loopCount;
//当前产品数量
int product;
//仓库能容纳的最大产品数量
int capacity;

QMutex mutex;
/*
The QWaitCondition class provides a condition variable for synchronizing threads
QWaitCondition类为线程同步提供了一个条件变量
*/
QWaitCondition productIsNotFull;
QWaitCondition productIsNotEmpty;

//生产者
class Producer : public QThread {
public:
Producer(MutexWaitCondition *manager) : QThread() {
this->manager = manager;
}
protected:
void run() {
for(int i=0; i<manager->loopCount; i++) {
manager->mutex.lock();
while(manager->product == manager->capacity) {
/*
bool QWaitCondition::wait(QReadWriteLock * lockedReadWriteLock, unsigned long time = ULONG_MAX)
Releases the lockedReadWriteLock and waits on the wait condition
释放该锁，并且阻塞当前线程，直到条件满足（即调用wake方法被唤醒）
*/
manager->productIsNotFull.wait(&manager->mutex);
}
manager->product++;
//cout<<"P";
cout<<i<<".P="<<manager->product<<", ";
manager->productIsNotEmpty.wakeAll();
manager->mutex.unlock();
}
}
private:
MutexWaitCondition *manager;
};

//消费者
class Consumer : public QThread {
public:
Consumer(MutexWaitCondition *manager) : QThread() {
this->manager = manager;
}
protected:
void run() {
for(int i=0; i<manager->loopCount; i++) {
manager->mutex.lock();
while(manager->product == 0) {
manager->productIsNotEmpty.wait(&manager->mutex);
}
manager->product--;
//cout<<"C";
cout<<i<<".C="<<manager->product<<", ";
manager->productIsNotFull.wakeAll();
manager->mutex.unlock();
}
}
private:
MutexWaitCondition *manager;
};

//无修饰的方法，默认是private的
public:
void test(int loopCount, int capacity)
{
this->loopCount = loopCount;
this->capacity = capacity;
this->product = 0;

Producer producer(this);
Consumer consumer(this);
//thread.start会调用thread内部的run方法
producer.start();
consumer.start();
/*
Blocks the thread until either of these conditions is met:
阻塞该线程直到所有条件都满足
*/
producer.wait();
consumer.wait();
cout<<endl;
}
};
#endif // MUTEXWAITCONDITION

（3）SemaphoreMutex.h

#ifndef SEMAPHOREMUTEX
#define SEMAPHOREMUTEX

#include <QCoreApplication>
#include <iostream>
#include <QThread>
#include <QSemaphore>
//注意:网上很多Semaphore不用mutex的做法都是错误的
#include <QMutex>

using namespace std;

class SemaphoreMutex {
public:
//预计生产(或消费)数量
int loopCount;
//当前产品数量: productSemphore.avaliable()
//int product;
//仓库能容纳的最大产品数量
int capacity;

QMutex mutex;
/*
The QSemaphore class provides a general counting semaphore
QSemaphore类提供了一个通用的计数信号量
*/
QSemaphore *productSemphore;
QSemaphore *leftSpaceSemaphore;

//生产者
class Producer : public QThread {
public:
Producer(SemaphoreMutex *manager) : QThread() {
this->manager = manager;
}
protected:
void run() {
for(int i=0; i<manager->loopCount; i++) {
/*
Tries to acquire n resources guarded by the semaphore. If n > available(), this call will block until enough resources are available.
尝试去获取(减去)n个被信号量控制的资源。如果n>可用资源数量，它就会阻塞直到有足够的资源为止。
*/
manager->leftSpaceSemaphore->acquire();
//之所以lock要在acquire后面是因为: 如果消费者拿到了锁，那么又没有商品，那么久会导致死锁
manager->mutex.lock();
manager->productSemphore->release();
//cout<<"P";
cout<<i<<".P="<<manager->productSemphore->available()<<", ";
manager->mutex.unlock();
}
}
private:
SemaphoreMutex *manager;
};

//消费者
class Consumer : public QThread {
public:
Consumer(SemaphoreMutex *manager) : QThread() {
this->manager = manager;
}
protected:
void run() {
for(int i=0; i<manager->loopCount; i++) {
manager->productSemphore->acquire();
manager->mutex.lock();
manager->leftSpaceSemaphore->release();
//cout<<"C";
cout<<i<<".C="<<manager->productSemphore->available()<<", ";
manager->mutex.unlock();
}
}
private:
SemaphoreMutex *manager;
};

//无修饰的方法，默认是private的
public:
void test(int loopCount, int capacity)
{
this->loopCount = loopCount;
this->capacity = capacity;

//参数为: 信号量的当前值
productSemphore = new QSemaphore(0);
leftSpaceSemaphore = new QSemaphore(capacity);

Producer producer(this);
Consumer consumer(this);
//thread.start会调用thread内部的run方法
producer.start();
consumer.start();
/*
Blocks the thread until either of these conditions is met:
阻塞该线程直到所有条件都满足
*/
producer.wait();
consumer.wait();
cout<<endl;
}
};

#endif // SEMAPHOREMUTEX

（4）main.cpp

#include <QCoreApplication>
#include "mutexonly.h"
#include "mutexwaitcondition.h"
#include "semaphoremutex.h"
#include <QTime>

int main(int argc, char *argv[])
{
int loopCount = 100;
int capacity = 3;
QTime time;
int mutexOnlyElapsed;
int mutexWaitConditionElapsed;
int SemaphoreMutexElapsed;

cout<<"loopCount = "<<loopCount<<", capacity = "<<capacity<<endl<<endl;

cout<<"MutexOnly"<<endl;
MutexOnly mutexOnly;
time.start();
mutexOnly.test(loopCount, capacity);
//milliseconds
mutexOnlyElapsed = time.elapsed();
cout<<"elapsed "<<mutexOnlyElapsed<<"ms"<<endl;
cout<<endl;

cout<<"MutexWaitCondition"<<endl;
MutexWaitCondition mutexWaitCondition;
time.restart();
mutexWaitCondition.test(loopCount, capacity);
mutexWaitConditionElapsed = time.elapsed();
cout<<"elapsed "<<mutexWaitConditionElapsed<<"ms"<<endl;
cout<<endl;

cout<<"SemaphoreMutex"<<endl;
SemaphoreMutex semaphoreMutex;
time.restart();
semaphoreMutex.test(loopCount, capacity);
SemaphoreMutexElapsed = time.elapsed();
cout<<"elapsed "<<SemaphoreMutexElapsed<<"ms"<<endl;
cout<<endl;
return 0;
}

运行截图

（1）当loopCount=100，capacity=3时

mutex method	elapsed time
MutexOnly	92ms
MutexWaitCondition	101ms
SemaphoreMutex	92ms

（2）当loopCount=10000，capacity=3时

mutex method	elapsed time
MutexOnly	7744ms
MutexWaitCondition	8756ms
SemaphoreMutex	8237ms

由此可见：mutex会稍微快一点，但是mutex也存在着一定的问题。因此，Qt为我们提供了QMutexLocker类来封装QMutex。究竟是什么问题呢？且听下回分解。