c++11 线程池实现以及示例

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/haluoluo211/article/details/79659864

线程池的使用在工作中非常普遍,对于java,python而言线程池使用还是比较方便。去年年底由于工作需要,用c++搭建一套工程代码,需要使用线程池,但是C++中并没有现有的线程池,为了快速开发,以及代码的稳定还是google在github上面找到了一个不错的C++11实现的版本,然后做了一点修改(兼容服务器gcc4.7)。

通过这一个线程池的实现,收获如下:

1. 更加深入的理解了线程池
2. 熟悉了C++11的,模板,mutex, condition_variable, std::function, std::bind, std::unique_lock等的使用

下面首先给出一个简单的应用场景,阻塞等待请求,或者事件(例如epoll事件),然后将数据以及handle压入线程池处理,这样无需等待handle函数处理完,即可等待下一次的请求或者事件(相当于是异步)。

    //创建一个还有5个线程的线程池
    ThreadPool tp(5);
    while(true){
        // 阻塞的等待http的请求数据data
        data = wait_req_from_http();
        //将数据压入线程池中处理,其中process_data_function为处理data的函数
        tp.enqueue(data, process_data_function);
    }

下面给出完整代码,只有一个.h头文件,这样方便工程的搭建:

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>

class ThreadPool {
public:
    ThreadPool(size_t);
    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args)
    -> std::future<typename std::result_of<F(Args...)>::type>;
    ~ThreadPool();
private:
    // 线程池
    std::vector< std::thread > workers;
    // 任务队列
    std::queue< std::function<void()> > tasks;

    // synchronization同步
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

// the constructor just launches some amount of workers
inline ThreadPool::ThreadPool(size_t threads)
        :   stop(false)
{
    //创建n个线程,每个线程等待是否有新的task, 或者线程stop(要终止)
    for(size_t i = 0;i<threads;++i)
        workers.emplace_back(
                [this]
                {
                    for(;;)//轮询
                    {
                        std::function<void()> task;

                        {
                        //获取同步锁
                                                    std::unique_lock<std::mutex> lock(this->queue_mutex);
                                                    //线程会一直阻塞,直到有新的task,或者是线程要退出
                            this->condition.wait(lock,
                                                 [this]{ return this->stop || !this->tasks.empty(); });
                                                 //线程退出
                            if(this->stop && this->tasks.empty())
                                return;
                                //将task取出
                            task = std::move(this->tasks.front());
                            //队列中移除以及取出的task
                            this->tasks.pop();
                        }
                        //执行task,完了则进入下一次循环
                        task();
                    }
                }
        );
}

// add new work item to the pool
// 将队列压入线程池,其中f是要执行的函数, args是多有的参数
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type>
{
    //返回的结果的类型,当然可以根据实际的需要去掉这个(gcc4.7的c++11不支持)
    using return_type = typename std::result_of<F(Args...)>::type;
//将函数handle与参数绑定
    auto task = std::make_shared< std::packaged_task<return_type()> >(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
    );

    //after finishing the task, then get result by res.get() (mainly used in the invoked function)
    std::future<return_type> res = task->get_future();
    {
        //压入队列需要线程安全,因此需要先获取锁
        std::unique_lock<std::mutex> lock(queue_mutex);

        // don't allow enqueueing after stopping the pool
        if(stop)
            throw std::runtime_error("enqueue on stopped ThreadPool");
        //任务压入队列
        tasks.emplace([task](){ (*task)(); });
    }
    //添加了新的task,因此条件变量通知其他线程可以获取task执行
    condition.notify_one();
    return res;
}

// the destructor joins all threads
inline ThreadPool::~ThreadPool()
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();/通知所有在等待锁的线程
    //等待所有的线程任务执行完成退出
    for(std::thread &worker: workers)
        worker.join();
}

下面给出运行的示例以及结果(推荐使用CLion):

//线程池要执行的任务

    //模拟下载,sleep 2S
    void dummy_download(std::string url){
        std::this_thread::sleep_for(std::chrono::seconds(3));
        std::cout<<" complete download: "<<url<<std::endl;
    }

    //根据id返回用户名
    std::string get_user_name(int id){
        std::this_thread::sleep_for(std::chrono::seconds(2));
        return "user_" + std::to_string(id);
    }

线程池使用示例:

        ThreadPool tp(5);

        //线程执行结果的返回
        std::vector<std::future<std::string>> vecStr;
        // 下载对应的url链接,没有返回值
        tp.enqueue(dummy_download, "www.baidu.jpg");
        tp.enqueue(dummy_download, "www.yy.jpg");

        //数据库根据id查询user name
        vecStr.emplace_back(tp.enqueue(get_user_name, 101));
        vecStr.emplace_back(tp.enqueue(get_user_name, 102));

        //输出线程返回的值,实际中可以不要
        std::future<void> res1 = std::async(std::launch::async, [&vecStr](){
            for (auto &&ret:vecStr) {
                std::cout<<"get user: "<<ret.get();
            }
            std::cout<<std::endl;
        });

运行结果如下:

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏web编程技术分享

【Java框架型项目从入门到装逼】第五节 - 在Servlet中接收和返回数据

35870
来自专栏Vamei实验室

Python标准库03 路径与文件 (os.path包, glob包)

路径与文件的简介请参看Linux文件系统 os.path包 os.path包主要是处理路径字符串,比如说'/home/vamei/doc/file.txt',提...

26780
来自专栏蓝天

为什么ps中CPU占用率会有超出%100的现象?

前面的关于ps中的%CPU的含义一文已经介绍了CPU占用率的含义,那么为什么有时会在ps的输出中看到CPU占用率超出%100的现象呢?我们知道在/proc目录下...

13920
来自专栏IT派

Python 的异步 IO:Asyncio 简介

所谓「异步 IO」,就是你发起一个 IO 操作,却不用等它结束,你可以继续做其他事情,当它结束时,你会得到通知。

13930
来自专栏应兆康的专栏

Python Web - Flask笔记8

CSRF(Cross Site Request Forgery)跨站域请求伪造是一种网络攻击方式。

12610
来自专栏运维小白

Linux基础(day18)

5.5 进入编辑模式 编辑模式:就是进入到可以编辑文本文件的模式 ? 进入编辑模式方法: i ,直接在当前光标处进入编辑模式 I (大写I),光标直接移动到...

18970
来自专栏闻道于事

Linux-看完这篇Linux基本的操作就会了

33930
来自专栏北京马哥教育

Shell的18条常用命令整理

? 作者 | junmail 来源 | CSDN社区 云豆贴心提醒,本文阅读时间8分钟,文末有秘密! 1、 ls: 类似于dos下的dir命令 ...

41140
来自专栏咸鱼不闲

post和get的区别,面试经常被问到!(二)

39820
来自专栏happyJared

Linux私房菜:文件目录管理及常见命令

10010

扫码关注云+社区

领取腾讯云代金券