条件变量pthread_cond_t又名管程，条件变量的正确使用方式：
1 必须与pthread_mutex_t一起使用，条件的读写受互斥量保护
2 必须在pthread_mutex_lock后才能调用pthread_cond_wait
3 采用while抱住pthread_cond_wait，这样防止虚假唤醒(线程甚至在没有其它向条件变量发送信号的情况下就有可能会被唤醒)
阻塞队列：
BlockingQueue:阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，往队列里添加元素的操作会被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样，试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列。
CountDownLatch:CountDownLatch类是一个同步倒数计数器,构造时传入int参数,该参数就是计数器的初始值，每调用一次countDown()方法，计数器减1,计数器大于0 时，await()方法会阻塞后面程序执行，直到计数器为0
- #include<iostream>
- #include<assert.h>
- #include<pthread.h>
- #include<unistd.h>
- #include<boost/noncopyable.hpp>
- #include<deque>
- using namespace std;
- using namespace boost;
- class Mutex:public noncopyable{//互斥量的封装
- public:
- Mutex(){
- pthread_mutex_init(&mutex,NULL);
- }
- void lock(){
- pthread_mutex_lock(&mutex);
- }
- void unlock(){
- pthread_mutex_unlock(&mutex);
- }
- ~Mutex(){
- pthread_mutex_destroy(&mutex);
- }
- pthread_mutex_t* getMutex(){
- return &mutex;
- }
- private:
- mutable pthread_mutex_t mutex;
- };
- class MutexLockGuard:noncopyable{//RAII管理互斥量
- public:
- explicit MutexLockGuard(Mutex& mutex):mutex_(mutex){
- mutex_.lock();
- }
- ~MutexLockGuard(){
- mutex_.unlock();
- }
- private:
- Mutex& mutex_;//注意是引用，Mutex继承了noncopyable后不能拷贝构造
- };
- class Condition:public noncopyable{//条件变量的封装
- public:
- explicit Condition(Mutex& mutex):mutex_(mutex){
- pthread_cond_init(&cond,NULL);
- }
- ~Condition(){
- pthread_cond_destroy(&cond);
- }
- void wait(){
- pthread_cond_wait(&cond,mutex_.getMutex());
- }
- void notify(){
- pthread_cond_signal(&cond);
- }
- void notifyALL(){
- pthread_cond_broadcast(&cond);
- }
- private:
- Mutex& mutex_;//注意是引用
- pthread_cond_t cond;
- };
- template<typename T>//队列任务T
- class BlockingQueue:noncopyable{//无限容量的阻塞队列
- public:
- BlockingQueue():mutex(),notEmpty(mutex),Q(){}
- void put(const T& x){//向队列添加任务T
- MutexLockGuard guard(mutex);//栈对象
- Q.push_back(x);
- notEmpty.notifyALL();//唤醒等待条件上的线程
- cout<<"put()"<<endl;
- }
- T take(){//获取一个队列任务
- MutexLockGuard guard(mutex);
- while(Q.empty()){//循环抱住以防止假唤醒
- cout<<"Q.empty()"<<endl;
- notEmpty.wait();//原子的解锁mutex并进入等待，不会与put死锁
- }
- assert(!Q.empty());
- T front(Q.front());//这里假设T允许拷贝构造
- Q.pop_front();
- cout<<"take()"<<endl;
- return front;
- }
- size_t size() const{//返回队列大小
- MutexLockGuard guard(mutex);
- return Q.size();
- }
- private:
- mutable Mutex mutex;//互斥锁保护队列
- Condition notEmpty;//条件变量
- deque<T> Q;
- };
- class test{//任务T类型
- public:
- void show(){
- cout<<"show()"<<endl;
- }
- };
- BlockingQueue<test*> bq;
- void* worker1(void* arg){//线程1,添加一个T到任务队列
- sleep(2);
- test* temp=(test*)arg;
- bq.put(temp);
- }
- void* worker2(void* arg){//线程2,从任务队列获取一个T
- //sleep(1);
- test* temp=bq.take();
- temp->show();
- }
- int main(){
- pthread_t pid1,pid2;
- test* temp=new test;
- pthread_create(&pid1,NULL,worker1,&temp);
- pthread_create(&pid2,NULL,worker2,NULL);
- pthread_join(pid1,NULL);
- pthread_join(pid2,NULL);
- delete temp;
- return 0;
- }
输出：
Q.empty()
put()
take()
show()
采用CoutDownLatch作为同步线程间工具：
- #include<iostream>
- #include<assert.h>
- #include<pthread.h>
- #include<unistd.h>
- #include<boost/noncopyable.hpp>
- #include<deque>
- using namespace std;
- using namespace boost;
- class Mutex:public noncopyable{//互斥量的封装
- public:
- Mutex(){
- pthread_mutex_init(&mutex,NULL);
- }
- void lock(){
- pthread_mutex_lock(&mutex);
- }
- void unlock(){
- pthread_mutex_unlock(&mutex);
- }
- ~Mutex(){
- pthread_mutex_destroy(&mutex);
- }
- pthread_mutex_t* getMutex(){
- return &mutex;
- }
- private:
- mutable pthread_mutex_t mutex;
- };
- class MutexLockGuard:noncopyable{//RAII管理互斥量
- public:
- explicit MutexLockGuard(Mutex& mutex):mutex_(mutex){
- mutex_.lock();
- }
- ~MutexLockGuard(){
- mutex_.unlock();
- }
- private:
- Mutex& mutex_;//注意是引用，Mutex继承了noncopyable后不能拷贝构造
- };
- class Condition:public noncopyable{//条件变量的封装
- public:
- explicit Condition(Mutex& mutex):mutex_(mutex){
- pthread_cond_init(&cond,NULL);
- }
- ~Condition(){
- pthread_cond_destroy(&cond);
- }
- void wait(){
- pthread_cond_wait(&cond,mutex_.getMutex());
- }
- void notify(){
- pthread_cond_signal(&cond);
- }
- void notifyALL(){
- pthread_cond_broadcast(&cond);
- }
- private:
- Mutex& mutex_;//注意是引用
- pthread_cond_t cond;
- };
- template<typename T>//队列任务T
- class BlockingQueue:noncopyable{//无限容量的阻塞队列
- public:
- BlockingQueue():mutex(),notEmpty(mutex),Q(){}
- void put(const T& x){//向队列添加任务T
- MutexLockGuard guard(mutex);//栈对象
- Q.push_back(x);
- notEmpty.notifyALL();//唤醒等待条件上的线程
- cout<<"put()"<<endl;
- }
- T take(){//获取一个队列任务
- MutexLockGuard guard(mutex);
- while(Q.empty()){//循环抱住以防止假唤醒
- cout<<"Q.empty()"<<endl;
- notEmpty.wait();//原子的解锁mutex并进入等待，不会与put死锁
- }
- assert(!Q.empty());
- T front(Q.front());//这里假设T允许拷贝构造
- Q.pop_front();
- cout<<"take()"<<endl;
- return front;
- }
- size_t size() const{//返回队列大小
- MutexLockGuard guard(mutex);
- return Q.size();
- }
- private:
- mutable Mutex mutex;//互斥锁保护队列
- Condition notEmpty;//条件变量
- deque<T> Q;
- };
- class test{//任务T类型
- public:
- void show(){
- cout<<"show()"<<endl;
- }
- };
- class CountDownLatch:noncopyable{//
- public:
- explicit CountDownLatch(int count):mutex(),condition(mutex),cnt(count){}//mutex要先于condition构造
- void wait(){//阻塞等待计数为0
- MutexLockGuard guard(mutex);
- while(cnt>0){
- condition.wait();
- }
- }
- void countDown(){//将计数减1
- MutexLockGuard guard(mutex);
- --cnt;
- if(cnt==0){
- condition.notifyALL();//
- }
- }
- int getCount() const{
- MutexLockGuard guard(mutex);
- return cnt;
- }
- private:
- mutable Mutex mutex;
- Condition condition;
- int cnt;
- };
- BlockingQueue<test*> bq;
- CountDownLatch latch(1);//计数为1
- void* worker1(void* arg){//线程1,添加一个T到任务队列
- //sleep(2);
- latch.countDown();//线程1将计数减一
- test* temp=(test*)arg;
- bq.put(temp);
- }
- void* worker2(void* arg){//线程2,从任务队列获取一个T
- //sleep(1);
- latch.wait();//线程2等待计数为0
- test* temp=bq.take();
- temp->show();
- }
- int main(){
- pthread_t pid1,pid2;
- test* temp=new test;
- pthread_create(&pid1,NULL,worker1,&temp);
- pthread_create(&pid2,NULL,worker2,NULL);
- pthread_join(pid1,NULL);
- pthread_join(pid2,NULL);
- delete temp;
- return 0;
- }
程序输出：
put()
take()
show()