生产者消费者模型（代码实现）

一个场所： 队列

两个角色： 生产者、消费者

三种关系：
生产者-生产者：互斥
消费者-消费者：互斥
生产者-消费者：同步+互斥

如果生产者直接将数据交给消费，那么代码的耦合性将会非常强，而且如果生产数据特别快而消耗数据特别慢将会造成数据堆积，数据可能会丢失
因此利用 场所，生产者将数据放到场所中，消费者从场所中拿取数据

场所的作用：
解耦合
支持忙闲不均
支持并发

条件变量实现

/*==============================================================
*    Copyright . All rights reserved.
*    Filename:  pro_con.cpp
*    Author:    shen
*    Last modified: 2019-06-26 16:57
*    Description: 条件变量实现生产者消费者模型
==============================================================*/
#include <cstdio>
#include <queue>
#include <unistd.h>
#include <pthread.h>

#define MAX_QUEUE 5

class BlockQueue {
    public:
        BlockQueue() {
            pthread_mutex_init(&_mutex, NULL);
            pthread_cond_init(&_cond_pro, NULL);
            pthread_cond_init(&_cond_con, NULL);
        }
        ~BlockQueue() {
            pthread_mutex_destroy(&_mutex);
            pthread_cond_destroy(&_cond_pro);
            pthread_cond_destroy(&_cond_con);
        }
        void Push(int data) {
            //加锁,队列满了则等待,生产数据,解锁
            pthread_mutex_lock(&_mutex);
            while (_queue.size() == MAX_QUEUE) {
                pthread_cond_wait(&_cond_pro, &_mutex);
            }
            _queue.push(data);
            pthread_cond_signal(&_cond_con);
            pthread_mutex_unlock(&_mutex);
        }
        //data 输出型参数,返回删除的数据
        void Pop(int *data) {
        	//加锁,队列为空则等待,消费数据,解锁
            pthread_mutex_lock(&_mutex);
            while (_queue.empty()) {
                pthread_cond_wait(&_cond_con, &_mutex);
            }
            *data = _queue.front();
            _queue.pop();
            pthread_cond_signal(&_cond_pro);
            pthread_mutex_unlock(&_mutex);
        }
        int Size() {
            return _queue.size();
        }
    private:
        std::queue<int> _queue; 
        pthread_mutex_t _mutex;
        pthread_cond_t _cond_pro;
        pthread_cond_t _cond_con;
};
//保证资源++是原子操作(多个生产者同时生产数据)
pthread_mutex_t mut; 
int resources = 0;
void* thr_pro(void* arg) {
    BlockQueue* q = (BlockQueue*)arg;

    while (1) {
        pthread_mutex_lock(&mut);
        q->Push(resources++);
        pthread_mutex_unlock(&mut);
        printf("   %lu product %04d %6d\n", pthread_self(), resources, q->Size());
        usleep(200000);
    }

    return NULL; 
}

void* thr_con(void* arg) {
    BlockQueue* q = (BlockQueue*)arg;

    while (1) {
        int data;
        q->Pop(&data);
        printf("%lu consume %04d %6d\n", pthread_self(), data, q->Size());
        usleep(200000);
    }

    return NULL;
}

int main() {
    BlockQueue* q = new BlockQueue;
    pthread_mutex_init(&mut, NULL);

    //四个消费者四个生产者
    pthread_t pro_tid[4], con_tid[4];
    for (int i = 0; i < 4; ++i) {
        int ret = pthread_create(&pro_tid[i], NULL, thr_pro, (void*)q);
        if (ret != 0) {
            perror("pthread_create error~~\n");
            return -1;
        }
        ret = pthread_create(&con_tid[i], NULL, thr_con, (void*)q);
        if (ret != 0) {
            perror("pthread_create error~~\n");
            return -1;
        }
    }

    for (int i = 0; i < 4; ++i) {
        pthread_join(pro_tid[i], NULL);
        pthread_join(con_tid[i], NULL);
    }
    pthread_mutex_destroy(&mut);

    return 0;
}

信号量实现

/*==============================================================
*    Copyright . All rights reserved.
*    Filename:  sem.cpp
*    Author:    shen
*    Last modified: 2019-06-26 16:57
*    Description: 信号量实现生产者消费者模型
==============================================================*/
#include <vector>
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>

#define MAX_QUEUE 5

class RingQueue {
    public:
        RingQueue(int cap = MAX_QUEUE)
            : _queue(5)
            , _front(0)
            , _rear(0)
            , _capacity(cap)
        {
            sem_init(&_sem_lock, 0, 1); //互斥锁只有 0 和 1
            sem_init(&_sem_data, 0, 0);
            sem_init(&_sem_space, 0, _capacity);
        }
        ~RingQueue() {
            sem_destroy(&_sem_lock);
            sem_destroy(&_sem_data);
            sem_destroy(&_sem_space);
        }
        void Push(int data) {
            sem_wait(&_sem_space);
            //先等待空闲空间,有空闲空间才能进行加锁并且对资源进行操作
            //如果先加锁,然后没有空间资源,陷入等待将程序卡死
            sem_wait(&_sem_lock);
            _queue[_front] = data;
            _front = (_front + 1) % _capacity;
            sem_post(&_sem_lock);
            sem_post(&_sem_data);
        }
        void Pop(int *data) {
            sem_wait(&_sem_data);
            sem_wait(&_sem_lock);
            *data = _queue[_rear];
            _rear = (_rear + 1) % _capacity;
            sem_post(&_sem_lock);
            sem_post(&_sem_space);
        }
        int Size() {
            int size = 0;
            int front = _front;
            while (front != _rear) {
                front = (front + 1) % _capacity;
                size++;
            }
            return size;
        }
    private:
        std::vector<int> _queue;
        int _front;
        int _rear;
        int _capacity;
        sem_t _sem_lock;
        sem_t _sem_data;
        sem_t _sem_space;
};

int resources = 0;
//这个锁确保同一时刻只有一个线程进行资源++操作
//因为一旦一个进程拿着数据进入到Push函数中,还没有加锁(意味着也没有进行资源++)
//时间片轮转到了另一个线程,这个线拿着同样的数据(因为刚刚的线程还没有进行++操作),此时这个线程进行了++操作
//这种情况下,同样数据被加载了两次,因此需要保证Push的操作原子性
sem_t sem_lock;

void* thr_pro(void* arg) {
    RingQueue* q = (RingQueue*)arg;

    while (1) {
        sem_wait(&sem_lock);
        q->Push(resources++);
        sem_post(&sem_lock);
        printf("   %lu product %04d %6d\n", pthread_self(), resources, q->Size());
        usleep(10000);
    }

    return NULL; 
};

void* thr_con(void* arg) {
    RingQueue* q = (RingQueue*)arg;

    while (1) {
        int data;
        q->Pop(&data);
        printf("%lu consume %04d %6d\n", pthread_self(), data, q->Size());
        usleep(10000);
    }

    return NULL;
}

int main() {
    RingQueue* q = new RingQueue;
    sem_init(&sem_lock, 0, 1);

    //四个消费者四个生产者
    pthread_t pro_tid[4], con_tid[4];
    for (int i = 0; i < 4; ++i) {
        int ret = pthread_create(&pro_tid[i], NULL, thr_pro, (void*)q);
        if (ret != 0) {
            perror("pthread_create error~~\n");
            return -1;
        }
        ret = pthread_create(&con_tid[i], NULL, thr_con, (void*)q);
        if (ret != 0) {
            perror("pthread_create error~~\n");
            return -1;
        }
    }

    for (int i = 0; i < 4; ++i) {
        pthread_join(pro_tid[i], NULL);
        pthread_join(con_tid[i], NULL);
    }
    sem_destroy(&sem_lock);

    return 0;
}