基于c++11的短小精悍生产者消费者模型

传统的c++98编写起来代码量挺大的，且受约束和限制，所以在空闲时间自写了个基于c++11的生产者消费者模型，采用了boost部分库，代码量一下子变得很简短且清晰。

#include "stdafx.h"
#include <iostream>
#include <queue>
#include <mutex>
#include <vector>
#include <thread>
#include <boost/any.hpp> 
#include <boost/thread.hpp>
#include <boost/lexical_cast.hpp>

class task_handler
{
public:
	task_handler() {}
	~task_handler() 
	{
		end_handle();
	}

	int init(const std::function<void(const boost::any&)>& f)
	{
		consumer_threadptr_ = std::make_shared<boost::thread>(&task_handler::handle_task, this);
		if (!consumer_threadptr_)
		{
			return -1;
		}

		f_ = f;
		return 0;
	}

	void add_task(const boost::any& t, bool tail=true)
	{
		std::unique_lock<std::mutex> lock(mux_);
		if (tail)
		{
			que_.push_back(t);
		}
		else
		{
			que_.push_front(t);
		}
		cond_.notify_all();
	}

	void end_handle()
	{
		if (consumer_threadptr_ && consumer_threadptr_->joinable())
		{
			consumer_threadptr_->interrupt();
			consumer_threadptr_->join();
		}
	}

protected:
	void handle_task()
	{
		while (true)
		{
			std::unique_lock<std::mutex> lock(mux_);
			while (que_.empty())
			{
				cond_.wait(lock); 
			}

			while (!que_.empty())
			{
				boost::any& t = que_.front();
				f_(t);
				que_.pop_front();
			}
		}
	}

private:
	std::deque<boost::any> que_;
	std::mutex mux_;
	boost::condition_variable_any cond_;
	std::function<void(const boost::any&)> f_;
	std::shared_ptr<boost::thread> consumer_threadptr_;
};

typedef struct  
{
	uint32_t id;
	std::string name;
} message;

int main()
{
	auto f = [](const boost::any& data)
	{
		if (data.type() == typeid(int))
		{
			std::cout << "type=int, value=" << boost::any_cast<int>(data) << std::endl;
		}
		else if (data.type() == typeid(std::string))
		{
			std::cout << "type=string, value=" << (boost::any_cast<std::string>(data)).c_str() << std::endl;
		}
		else if (data.type() == typeid(message))
		{
			std::cout << "type=message, id=" << (boost::any_cast<message>(data)).id
				<< ", name=" << (boost::any_cast<message>(data)).name.c_str() << std::endl;
		}
	};

	task_handler ther;
	ther.init(f);
	std::string str = "string";
	message msg;
	for (auto i = 0; i < 100; i++)
	{
		switch (i % 3)
		{
		case 0:
		{
			ther.add_task(i);
		}
		break;
		case 1:
		{
			std::string temp = str + boost::lexical_cast<std::string>(i);
			ther.add_task(temp);
		}
		break;
		case 2:
		{
			msg.id = i;
			msg.name = std::string("message_name") + boost::lexical_cast<std::string>(i);
			ther.add_task(msg, false);
		}
		break;
		}
	}

	getchar();
	return 0;
}

上述只是单生产单消费者模型，而且当某一任务耗时较长时会阻碍后续的任务处理，其实还可以通过修改构成单生产者多消费者模型。