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

#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;
}

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