muduo之ThreadPool

         muduo中的线程池。

ThreadPool.h

// Use of this source code is governed by a BSD-style license // that can be found in the License file. // // Author: Shuo Chen (chenshuo at chenshuo dot com)#ifndef MUDUO_BASE_THREADPOOL_H #define MUDUO_BASE_THREADPOOL_H#include "muduo/base/Condition.h" #include "muduo/base/Mutex.h" #include "muduo/base/Thread.h" #include "muduo/base/Types.h"#include <deque> #include <vector>namespace muduo {class ThreadPool : noncopyable {public:typedef std::function<void ()> Task;explicit ThreadPool(const string& nameArg = string("ThreadPool"));~ThreadPool();// Must be called before start().void setMaxQueueSize(int maxSize) { maxQueueSize_ = maxSize; }//设置任务队列的最大值void setThreadInitCallback(const Task& cb)//设置线程执行前的回调函数{ threadInitCallback_ = cb; }void start(int numThreads);void stop();const string& name() const{ return name_; }size_t queueSize() const;// Could block if maxQueueSize > 0// There is no move-only version of std::function in C++ as of C++14.// So we don't need to overload a const& and an && versions// as we do in (Bounded)BlockingQueue.// https://stackoverflow.com/a/25408989void run(Task f);private:bool isFull() const REQUIRES(mutex_);//判满void runInThread();//线程池的线程运行函数Task take();//取任务函数mutable MutexLock mutex_;Condition notEmpty_ GUARDED_BY(mutex_);//不空conditionCondition notFull_ GUARDED_BY(mutex_); //未满conditionstring name_;Task threadInitCallback_; //线程执行前的回调函数std::vector<std::unique_ptr<muduo::Thread>> threads_;//存放线程的队列std::deque<Task> queue_ GUARDED_BY(mutex_);//存放任务的队列,Task是function泛型size_t maxQueueSize_;//任务队列大小bool running_; };} // namespace muduo#endif // MUDUO_BASE_THREADPOOL_H

ThreadPool.cc

// Use of this source code is governed by a BSD-style license // that can be found in the License file. // // Author: Shuo Chen (chenshuo at chenshuo dot com)#include "muduo/base/ThreadPool.h"#include "muduo/base/Exception.h"#include <assert.h> #include <stdio.h> #include <iostream>using namespace muduo;ThreadPool::ThreadPool(const string& nameArg): mutex_(),notEmpty_(mutex_), //初始化的时候需要把condition和mutex关联起来notFull_(mutex_),name_(nameArg),maxQueueSize_(0),//初始化0running_(false) { }ThreadPool::~ThreadPool() {if (running_)//如果线程池在运行，那就要进行内存处理，在stop()函数中执行{stop();} }void ThreadPool::start(int numThreads) {assert(threads_.empty());running_ = true;threads_.reserve(numThreads);//预留reserver个空间for (int i = 0; i < numThreads; ++i){char id[32];//id存储线程idsnprintf(id, sizeof id, "%d", i+1);threads_.emplace_back(new muduo::Thread(std::bind(&ThreadPool::runInThread, this), name_+id));threads_[i]->start();//线程启动}if (numThreads == 0 && threadInitCallback_){threadInitCallback_();} }void ThreadPool::stop() //释放资源 {{MutexLockGuard lock(mutex_);running_ = false;notEmpty_.notifyAll();}for (auto& thr : threads_){thr->join();//资源回收} }size_t ThreadPool::queueSize() const //返回任务队列的大小 {MutexLockGuard lock(mutex_);return queue_.size(); }void ThreadPool::run(Task task)//将任务添加到任务队列 {printf("ThreadPool::run\n");if (threads_.empty()){task();}else{MutexLockGuard lock(mutex_);while (isFull())//判断任务队列是否满{printf("notFull_.wait\n");notFull_.wait();}assert(!isFull());printf("2222\n");queue_.push_back(std::move(task));notEmpty_.notify();//此时任务队列非空} }ThreadPool::Task ThreadPool::take()//取任务函数 {MutexLockGuard lock(mutex_);// always use a while-loop, due to spurious wakeupwhile (queue_.empty() && running_){notEmpty_.wait();}Task task;if (!queue_.empty()){task = queue_.front();queue_.pop_front(); if (maxQueueSize_ > 0) //取走任务后再判断任务队列的大小{notFull_.notify();}}return task; }bool ThreadPool::isFull() const {mutex_.assertLocked();std::cout << "maxQueueSize_" << maxQueueSize_ << "queue_.size()=" <<queue_.size() << std::endl;return maxQueueSize_ > 0 && queue_.size() >= maxQueueSize_; }void ThreadPool::runInThread()//线程池的线程运行函数 {try{if (threadInitCallback_){threadInitCallback_();}while (running_){Task task(take());if (task){task();}}}catch (const Exception& ex){fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());fprintf(stderr, "reason: %s\n", ex.what());fprintf(stderr, "stack trace: %s\n", ex.stackTrace());abort();}catch (const std::exception& ex){fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());fprintf(stderr, "reason: %s\n", ex.what());abort();}catch (...){fprintf(stderr, "unknown exception caught in ThreadPool %s\n", name_.c_str());throw; // rethrow} }

 

总结

以上是生活随笔为你收集整理的muduo之ThreadPool的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。