Linux C++ 使用condition实现阻塞队列的方法

/*
 * BlockingQueue.h
 *
 * Created on: 2014年6月10日
 *   Author: 
 */
#ifndef BLOCKINGQUEUE_H_
#define BLOCKINGQUEUE_H_
#include <iostream>
#include <pthread.h>
using namespace std;
//template <typename T >
class BlockingQueue
{
public:
	BlockingQueue();
	BlockingQueue(int capacity);
	~BlockingQueue();
	bool push(int item);
	int poll();
private:
	int capacity;
	int* queue;
	int head,tail;
	pthread_mutex_t mutex;
	pthread_cond_t notFull,notEmpty;
};

#endif /* BLOCKINGQUEUE_H_ */

/*
* BlockingQueue.cpp
*
* Created on: 2014年6月10日
* Author: 
*/
#include "../include/BlockingQueue.h"
BlockingQueue::BlockingQueue()
{
 this->capacity = 10;
 queue = new int[capacity];
 head = 0,tail = 0;
 pthread_mutex_init(&mutex,NULL);
 pthread_cond_init(&notFull,NULL);
 pthread_cond_init(&notEmpty,NULL);
}
BlockingQueue::BlockingQueue(int capacity)
{
 this->capacity = capacity;
 queue = new int[capacity];
 cout << "capacity " << sizeof(queue) << endl;
 head = 0,tail = 0;
 pthread_mutex_init(&mutex,NULL);
 pthread_cond_init(&notFull,NULL);
 pthread_cond_init(&notEmpty,NULL);
}
BlockingQueue::~BlockingQueue()
{
 this->capacity = 0;
 head = 0,tail = 0;
 delete queue;
 pthread_mutex_destroy(&mutex);
 pthread_cond_destroy(&notFull);
 pthread_cond_destroy(&notEmpty);
}
bool BlockingQueue::push(int item)
{
 pthread_mutex_lock(&mutex);
 cout << "you want push " << item << endl;
 while((head + 1) % capacity == tail)//is full
 {
  cout << "is full,wait..." << endl;
  // push wait
  pthread_cond_wait(&notFull,&mutex);
  cout << "not full,unlock" << endl;
 }
 {
  queue[head] = item;
  head = (head + 1) % capacity;
  cout << "push " << item << endl;
  //wake up poll thread
  pthread_cond_signal(&notEmpty);
  pthread_mutex_unlock(&mutex);
  return true;
 }
}
int BlockingQueue::poll()
{
 pthread_mutex_lock(&mutex);
 int ret = 0;
 while(head == tail) // is empty
 {
  cout << "is empty,wait..." << endl;
  //poll wait
  pthread_cond_wait(&notEmpty,&mutex);
  cout << "not empty,unlock..." << endl;
 }
 {
 ret = queue[tail];
 tail = (tail + 1) % capacity;
 cout << "take " << ret << endl;
 //wake up push thread
 pthread_cond_signal(&notFull);
 pthread_mutex_unlock(&mutex);
 return ret;
 }
}

#include <iostream>
#include "include/BlockingQueue.h"
using namespace std;
BlockingQueue queue(3);
void* put(void *)
{
	queue.push(1);
	  queue.push(2);
	  queue.push(3);
	  queue.push(4);
	  queue.push(5);
	  return NULL;
}
void* take(void *)
{
	queue.poll();
	queue.poll();
	queue.poll();
	return NULL;
}

int main() {
	pthread_t put1,take1;
  pthread_create(&put1,NULL,put,0);
  pthread_create(&take1,NULL,take,0);
  void * retval;
  pthread_join(put1,&retval);
  pthread_join(take1,&retval);
	return 0;
}