线程（三）生产者消费者模型+POSIX信号量

#include <iostream>
#include <queue>
#include <stdlib.h>
#include <pthread.h>
 
#define NUM 8
 
class BlockQueue{
    private:
        std::queue<int> q;
        int cap;
        pthread_mutex_t lock;
        pthread_cond_t full;
        pthread_cond_t empty;
 
    private:
        void LockQueue()
        {
            pthread_mutex_lock(&lock);
        }
        void UnLockQueue()
        {
            pthread_mutex_unlock(&lock);
        }
        void ProductWait()
        {
            pthread_cond_wait(&full, &lock);
        }
        void ConsumeWait()
        {
            pthread_cond_wait(&empty, &lock);
        }
        void NotifyProduct()
        {
            pthread_cond_signal(&full);
        }
        void NotifyConsume()
        {
            pthread_cond_signal(&empty);
        }
        bool IsEmpty()
        {
            return ( q.size() == 0 ? true : false );
        }
        bool IsFull()
        {
            return ( q.size() == cap ? true : false );
            }
 
    public:
        BlockQueue(int _cap = NUM):cap(_cap)
        {
            pthread_mutex_init(&lock, NULL);
            pthread_cond_init(&full, NULL);
            pthread_cond_init(&empty, NULL);
        }
        void PushData(const int &data)
        {
            LockQueue();
            while(IsFull()){
                NotifyConsume();
                std::cout << "queue full, notify consume data, product stop." << 
std::endl;
                ProductWait();
            }
            q.push(data);
            UnLockQueue();
        }
        void PopData(int &data)
        {
            LockQueue();
            while(IsEmpty()){
                NotifyProduct();
                std::cout << "queue empty, notify product data, consume stop." << 
std::endl;
                ConsumeWait();
            }
            data = q.front();
            q.pop();
            UnLockQueue();
        }
        ~BlockQueue()
        {
            pthread_mutex_destroy(&lock);
            pthread_cond_destroy(&full);
            pthread_cond_destroy(&empty);
        }
};
 
void *consumer(void *arg)
{
    BlockQueue *bqp = (BlockQueue*)arg;
    int data;
    for( ; ; ){
        bqp->PopData(data);
        std::cout << "Consume data done : " << data << std::endl;
    }
}
void *producter(void *arg)
{
    BlockQueue *bqp = (BlockQueue*)arg;
    srand((unsigned long)time(NULL));
    for( ; ; ){
        int data = rand() % 1024;
        bqp->PushData(data);
        std::cout << "Prodoct data done: " << data << std::endl;
    }
}
 
int main()
{
    BlockQueue bq;
    pthread_t c,p;
    pthread_create(&c, NULL, consumer, (void*)&bq);
    pthread_create(&p, NULL, producter, (void*)&bq);
 
    pthread_join(c, NULL);
    pthread_join(p, NULL);
    return 0;
}

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数：
    pshared:0表示线程间共享，非零表示进程间共享
    value：信号量初始值

int sem_destroy(sem_t *sem);

功能：等待信号量，会将信号量的值减1
int sem_wait(sem_t *sem);

功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。
int sem_post(sem_t *sem);

#include <iostream>
#include <vector>
#include <stdlib.h>
#include <semaphore.h>
#include <pthread.h>
 
#define NUM 16
 
class RingQueue{
    private:
        std::vector<int> q;
        int cap;
        sem_t data_sem;
        sem_t space_sem;
        int consume_step;
        int product_step;
 
    public:
        RingQueue(int _cap = NUM):q(_cap),cap(_cap)
        {
            sem_init(&data_sem, 0, 0);
            sem_init(&space_sem, 0, cap);
            consume_step = 0;
            product_step = 0;
        }
        void PutData(const int &data)
        {
            sem_wait(&space_sem); // P
            q[consume_step] = data;
            consume_step++;
            consume_step %= cap;
            sem_post(&data_sem); //V
        }
        void GetData(int &data)
        {
            sem_wait(&data_sem);
            data = q[product_step];
            product_step++;
            product_step %= cap;
            sem_post(&space_sem);
        }
        ~RingQueue()
        {
            sem_destroy(&data_sem);
            sem_destroy(&space_sem);
        }
};
void *consumer(void *arg)
{
    RingQueue *rqp = (RingQueue*)arg;
    int data;
    for( ; ; ){
        rqp->GetData(data);
        std::cout << "Consume data done : " << data << std::endl;
        sleep(1);
    }
}
 
//more faster
void *producter(void *arg)
{
    RingQueue *rqp = (RingQueue*)arg;
    srand((unsigned long)time(NULL));
    for( ; ; ){
        int data = rand() % 1024;
        rqp->PutData(data);
        std::cout << "Prodoct data done: " << data << std::endl;
    //  sleep(1);
    }
}
 
int main()
{
    RingQueue rq;
    pthread_t c,p;
    pthread_create(&c, NULL, consumer, (void*)&rq);
    pthread_create(&p, NULL, producter, (void*)&rq);
 
    pthread_join(c, NULL);
    pthread_join(p, NULL);
}