muduo_base程式碼剖析之ThreadPool執行緒池
阿新 • • 發佈:2018-11-26
1. 執行緒池
執行緒池的問題本質上也是生產者消費者模型問題
- 生產者生產產品的過程,實際上就是由程式設計師向任務佇列中新增任務的過程(需要程式設計師控制),實現程式碼見下:
1. print函式是程式設計師自己手動定義的任務函式
2. run(Task task)介面的功能是將task新增到任務佇列中,並喚醒等待任務的執行緒
pool.run(print); //該句話表示:程式設計師手動將print任務新增到任務佇列
for (int i = 0; i < 100; ++i)
{
char buf[32];
snprintf (buf, sizeof buf, "task %d", i);
pool.run(std::bind(printString, std::string(buf)));
}
- 消費者是執行緒池中的執行緒,(不需要程式設計師控制),當任務佇列中有任務時,將會喚醒執行緒池中的空閒執行緒
2. muduo執行緒池程式碼詳解:ThreadPool
// ThreadPool.h檔案
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
void run(Task f);
private:
bool isFull(); //任務佇列滿
void runInThread();
Task take();
mutable MutexLock mutex_;
Condition notEmpty_;
Condition notFull_;
string name_;
Task threadInitCallback_;
std::vector<std::unique_ptr<muduo::Thread>> threads_;
std::deque<Task> queue_;
size_t maxQueueSize_; //任務佇列的容量
bool running_;
};
// ThreadPool.cc檔案
#include <muduo/base/ThreadPool.h>
#include <muduo/base/Exception.h>
#include <assert.h>
#include <stdio.h>
using namespace muduo;
ThreadPool::ThreadPool(const string& nameArg)
: mutex_(),
notEmpty_(mutex_),
notFull_(mutex_),
name_(nameArg),
maxQueueSize_(0),
running_(false)
{
}
ThreadPool::~ThreadPool()
{
if (running_)
{
stop();
}
}
// 1. 建立numThreads個執行緒,並放入執行緒佇列
// 2. 呼叫threads_[i]->start(),使每個執行緒都啟動執行緒函式runInThread()
void ThreadPool::start(int numThreads)
{
assert(threads_.empty());
running_ = true;
threads_.reserve(numThreads);
for (int i = 0; i < numThreads; ++i)
{
char id[32];
snprintf(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::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
}
}
ThreadPool::Task ThreadPool::take()
{
MutexLockGuard lock(mutex_);
// always use a while-loop, due to spurious wakeup
while (queue_.empty() && running_)
{
notEmpty_.wait();
}
Task task;
if (!queue_.empty())
{
task = queue_.front();
queue_.pop_front();
if (maxQueueSize_ > 0)
{
notFull_.notify();
}
}
return task;
}
void ThreadPool::stop()
{
{
MutexLockGuard lock(mutex_);
running_ = false;
notEmpty_.notifyAll();//通知所有的等待執行緒
}
for (auto& thr : threads_) //等待所有的執行緒執行完任務後再退出
{
thr->join();
}
//等價於 for_each(thread_.begin(),thread_.end(),
// boost::bind(&muduo::Thread::join,_1));
}
size_t ThreadPool::queueSize() const
{
MutexLockGuard lock(mutex_);
return queue_.size();
}
//
void ThreadPool::run(Task task)
{
if (threads_.empty()) //建立了0個執行緒,即沒有執行緒池,只有一個主執行緒
{
task(); //主執行緒強制執行該任務
}
else// 如果執行緒池中有空閒執行緒,就將任務新增到任務佇列
{
MutexLockGuard lock(mutex_);
while (isFull())
{
notFull_.wait();
}
assert(!isFull());
queue_.push_back(std::move(task));
notEmpty_.notify();
}
}
bool ThreadPool::isFull() const //任務佇列滿了
{
mutex_.assertLocked();
return maxQueueSize_ > 0 && queue_.size() >= maxQueueSize_;
}
3. 執行緒池測試示例
#include <muduo/base/ThreadPool.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/base/CurrentThread.h>
#include <muduo/base/Logging.h>
#include <stdio.h>
#include <unistd.h> // usleep
void print()
{
printf("tid=%d\n", muduo::CurrentThread::tid());
}
void printString(const std::string& str)
{
LOG_INFO << str;
usleep(100*1000);
}
void test(int maxSize)
{
LOG_WARN << "Test ThreadPool with max queue size = " << maxSize;
muduo::ThreadPool pool("MainThreadPool");
pool.setMaxQueueSize(maxSize); //設定任務佇列的大小
pool.start(5); //建立5個執行緒,並啟動執行緒
LOG_WARN << "Adding";
//向執行緒池中新增自定義的無參的print任務,並喚醒執行緒池中的空閒執行緒
pool.run(print);
pool.run(print);
for (int i = 0; i < 100; ++i)
{
char buf[32];
snprintf(buf, sizeof buf, "task %d", i);
//向執行緒池中新增自定義的有引數的printString任務,並喚醒執行緒池中的空閒執行緒
pool.run(std::bind(printString, std::string(buf)));
}
LOG_WARN << "Done";
//向執行緒池中新增CountDownLatch類中的成員函式countDown(),並喚醒執行緒池中的空閒執行緒
muduo::CountDownLatch latch(1);
pool.run(std::bind(&muduo::CountDownLatch::countDown, &latch));
latch.wait();
pool.stop();
}
int main()
{
test(0);
test(1);
test(5);
test(10);
test(50);
}