c++11多线程生产者消费者问题

生产者-消费者（producer-consumer）问题，也称作有界缓冲区（bounded-buffer）问题，两个进程共享一个公共的固定大小的缓冲区。其中一个是生产者，用于将消息放入缓冲区；另外一个是消费者，用于从缓冲区中取出消息。问题出现在当缓冲区已经满了，而此时生产者还想向其中放入一个新的数据项的情形，其解决方法是让生产者此时进行休眠，等待消费者从缓冲区中取走了一个或者多个数据后再去唤醒它。同样地，当缓冲区已经空了，而消费者还想去取消息，此时也可以让消费者进行休眠，等待生产者放入一个或者多个数据时再唤醒它。

头文件：

#include<memory>

#include<chrono>

#include<iostream>

#include<thread>

#include<condition_variable>

#include<mutex>

#include<cstdlib>

#include<chrono>

代码：

#include "stdafx.h"

class loop_list {

public:

enum { kItemRepositorySize = 10, kItemsToProduce = 1000};

private:

struct ItemRepository {

std::condition_variable repo_not_empty;

std::condition_variable repo_not_full;

std::mutex mtx;

size_t write_position = 0;

size_t read_position = 0;

int item_buffer[kItemRepositorySize]{ 0 };

int cnt = 0;

};

typedef struct ItemRepository ItemRepository;

ItemRepository ir;

size_t _stop;	//count about pop numbers

public:

void  push(int  _item) {

std::unique_lock<std::mutex> lock(ir.mtx);

while ((ir.write_position+1)% kItemRepositorySize

== ir.read_position){

std::cout << "please pop..." << std::endl;

ir.repo_not_full.wait(lock);

}

ir.item_buffer[ir.write_position] = _item;

(ir.write_position)++;

if (ir.write_position >= kItemRepositorySize)

ir.write_position = 0;

lock.unlock();

ir.repo_not_empty.notify_all();
}

std::shared_ptr<int> pop(){

std::unique_lock<std::mutex> lock(ir.mtx);

while (ir.write_position == ir.read_position) {

std::cout << "please push..." << std::endl;

ir.repo_not_empty.wait(lock);

}

std::shared_ptr<int> _result = std::make_shared<int>(ir.item_buffer[ir.read_position]);

(ir.read_position)++;

if (ir.read_position >= kItemRepositorySize)

ir.read_position = 0;

lock.unlock();

ir.repo_not_full.notify_all();

return _result;

}

size_t _count() const {

return _stop;

}

public:

loop_list(int _count = kItemsToProduce):_stop(_count) {};

};

//生产者
void ProducerTask(loop_list& ir) {

for (int i = 1; i <= ir._count(); ++i) {

//	std::this_thread::sleep_for( std::chrono::seconds(2));

std::cout <<"the id : "<<std::this_thread::get_id()<< ". Produce the " << i << "^th item..." << std::endl;

ir.push(i);

}
}

//消费者
void ConsumerTask(loop_list& ir,int ProducerTasknumber) //ProducerTasknumber 生产者数目
{
static int _count = 0;

while (true) {

//	std::this_thread::sleep_for(std::chrono::seconds(1));

auto item = ir.pop();

std::cout << "		Consume the " << *item << "^th item" << std::endl;

if (++_count == ProducerTasknumber*ir._count()) break;
}
}

int main(){

loop_list i;

std::thread one(ProducerTask, std::ref(i));
std::thread two(ProducerTask, std::ref(i));
std::thread thr(ConsumerTask, std::ref(i),2);

one.join();
two.join();
thr.join();

system("pause");

return 0;

}

将缓存区封装成一个类，将所有基本操作赋该类，并决定锁的粒度。将生产者与消费者操作封装成函数。生产者消费者互相不知道对方的存在，每个方法只是与缓存区交互。这种写法，我们在操作时只要确定生产产品与消费产品的数目后，再对操作进行修改，就不会牵一发而动全身。具有好的封装性。

部分参考：

点击打开链接