基于环形队列（信号量 + 锁）的生产者消费者模型
ringQueue.hpp（线程安全的环形队列）
#include <iostream>
#include <cstdlib>
#include <unistd.h>
#include <vector>
#include "sem.hpp"
#include "Mutex.hpp"
using namespace std;
#define RINGQUEUESIZE 10
template <class T>
class RingQueue
{
public:
RingQueue(int num = RINGQUEUESIZE)
:num_(num),c_step(0),p_step(0),space_sem_(num),data_sem_(0)
{
ring_queue_.reserve(num);
}
void push(const T& in)
{
space_sem_.P();
lock_Guard mtx(mtx_);
ring_queue_[p_step++] = in;
p_step %= num_;
//cout<<"push"<<endl;
data_sem_.V();
}
T pop()
{
data_sem_.P();
lock_Guard mtx(mtx_);
T t = ring_queue_[c_step++];
c_step %= num_;
//cout<<"pop"<<endl;
space_sem_.V();
return t;
}
private:
vector<T> ring_queue_;
int num_;
int c_step;
int p_step;
Sem space_sem_;
Sem data_sem_;
Mutex mtx_;
};
sem.hpp （封装信号量）
#include <iostream>
#include <semaphore.h>
using namespace std;
class Sem
{
public:
Sem(int value)
{
sem_init(&sem_,0,value);
}
void P()
{
sem_wait(&sem_);
}
void V()
{
sem_post(&sem_);
}
~Sem()
{
sem_destroy(&sem_);
}
private:
sem_t sem_;
};
Mutex.hpp （封装锁）
#include <iostream>
#include <cstdlib>
class Mutex
{
public:
//Mutex(pthread_mutex_t* mutex)
//:mutex_(mutex)
Mutex()
{
pthread_mutex_init(&mutex_, nullptr);
}
void lock()
{
pthread_mutex_lock(&mutex_);
}
void unlock()
{
pthread_mutex_unlock(&mutex_);
}
pthread_mutex_t* getMutex()
{
return &mutex_;
}
~Mutex()
{
pthread_mutex_destroy(&mutex_);
}
private:
pthread_mutex_t mutex_;
};
class lock_Guard
{
public:
lock_Guard(Mutex& mutex)
:mutex_(mutex)
{
mutex_.lock();
}
~lock_Guard()
{
mutex_.unlock();
}
private:
Mutex& mutex_;
};
Cond.hpp （封装条件变量）
#include <iostream>
#include <cstdlib>
#include "Mutex.hpp"
class Condition_variable
{
public:
Condition_variable(Mutex& mtx)
:mtx_(mtx)
{
pthread_cond_init(&cond_,nullptr);
}
void Wait()
{
pthread_cond_wait(&cond_,mtx_.getMutex());
}
void notify_one()
{
pthread_cond_signal(&cond_);
}
void notify_all()
{
pthread_cond_broadcast(&cond_);
}
~Condition_variable()
{
pthread_cond_destroy(&cond_);
}
private:
pthread_cond_t cond_;
Mutex& mtx_;
};
main.cc
#include "ringQueue.hpp"
#include <pthread.h>
#define CUSTOMERSIZE 2
#define PRODUCTORSIZE 1
class Task
{
public:
Task(int x,int y)
:x_(x),y_(y)
{}
int operator()()
{
return x_ + y_;
}
int x_;
int y_;
};
struct MyThread
{
RingQueue<Task>* rg;
string name;
};
void* customer(void* args)
{
MyThread* mt = (MyThread*)args;
RingQueue<Task>* rg = mt->rg;
string name = mt->name;
//RingQueue<Task>* rg = (RingQueue<Task>*)args;
sleep(3);
while(true)
{
Task task = rg->pop();
int ans = task();
cout << name << ": " << task.x_ << " + " << task.y_ << " = " << ans << endl;
sleep(1);
}
return nullptr;
}
void* productor(void* args)
{
MyThread* mt = (MyThread*)args;
RingQueue<Task>* rg = mt->rg;
string name = mt->name;
//RingQueue<Task>* rg = (RingQueue<Task>*)args;
while(true)
{
int x = rand()%100+1;
usleep(100);
int y = rand()%100+1;
rg->push(Task(x,y));
cout << name << ": " << x << " + " << y << " = ?"<<endl;
}
return nullptr;
}
int main()
{
srand(time(NULL));
RingQueue<Task> rg(5);
MyThread c_mt[CUSTOMERSIZE];
MyThread p_mt[PRODUCTORSIZE];
pthread_t c[CUSTOMERSIZE];
pthread_t p[PRODUCTORSIZE];
for(int i=0;i<CUSTOMERSIZE;++i)
{
c_mt[i].rg = &rg;
string name = "customer ";
name += to_string(i+1);
c_mt[i].name=name;
pthread_create(c+i,nullptr,customer,(void*)(c_mt+i));
}
for(int i=0;i<PRODUCTORSIZE;++i)
{
p_mt[i].rg = &rg;
string name = "productor ";
name += to_string(i+1);
p_mt[i].name=name;
pthread_create(p+i,nullptr,productor,(void*)(p_mt+i));
}
for(int i=0;i<CUSTOMERSIZE;++i)
{
pthread_join(c[i],nullptr);
}
for(int i=0;i<PRODUCTORSIZE;++i)
{
pthread_join(p[i],nullptr);
}
return 0;
}
makefile
ringQueue:main.cc
g++ -o $@ $^ -std=c++11 -l pthread
.PHONY:clean
clean:
rm -f ringQueue
