快捷導(dǎo)航

C++實(shí)現(xiàn)線程池的簡單方法示例

更新時(shí)間：2020年05月06日 11:04:42 作者：十面埋伏但莫慌

這篇文章主要給大家介紹了關(guān)于C++實(shí)現(xiàn)線程池的簡單方法，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家學(xué)習(xí)或者使用C++具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面來一起學(xué)習(xí)學(xué)習(xí)吧

最近自己寫了一個(gè)線程池。

總的來說，線程池就是有一個(gè)任務(wù)隊(duì)列，一個(gè)線程隊(duì)列，線程隊(duì)列不斷地去取任務(wù)隊(duì)列中的任務(wù)來執(zhí)行，當(dāng)任務(wù)隊(duì)列中為空時(shí)，線程阻塞等待新的任務(wù)添加過來。

我是用queue來存放任務(wù)，vector存放thread*，然后用condition_variable 來設(shè)置線程阻塞和喚醒。

下面直接上代碼吧。

線程池類頭文件Thread_Pool.h

/********************************************
   線程池頭文件

  Author:十面埋伏但莫慌
  Time:2020/05/03

*********************************************/
#pragma once
#ifndef _THREAD_POOL_H_
#define _THREAD_POOL_H_
#include<thread>
#include<queue>
#include<mutex>
#include<atomic>
#include<vector>
#include<condition_variable>

typedef std::function<void()> Func;//定義線程執(zhí)行函數(shù)類型，方便后面編碼使用。
//任務(wù)類
class Task {
public:
 Task() {}
 ~Task() {}
 int push(Func func);//添加任務(wù);
 int getTaskNum();//獲得當(dāng)前隊(duì)列中的任務(wù)數(shù);
 Func pop();//取出待執(zhí)行的任務(wù);
public:
 std::mutex mx;//鎖;
private:
 
 std::queue<Func> tasks;//任務(wù)隊(duì)列
};
//線程池類
class Thread_Pool {
public:
 Thread_Pool() :IsStart(false) {}
 ~Thread_Pool();
 int addTasks(Func tasks);//添加任務(wù);
 void start();//開啟線程池;
 void stop();//關(guān)閉線程池;
 void run();//線程工作函數(shù);
 int getTaskNum();//獲得當(dāng)前隊(duì)列中的任務(wù)數(shù);
private:
 static const int maxThreadNum = 3;//最大線程數(shù)為3;

 std::condition_variable cond;//條件量;
 std::vector<std::thread*> threads;//線程向量;
 std::atomic<bool> IsStart;//原子變量，判斷線程池是否運(yùn)行;
 Task tasks;//任務(wù)變量;
};
#endif

然后是線程池類成員函數(shù)定義文件Thread_Pool.cpp

/********************************************
   線程池CPP文件

  Author:十面埋伏但莫慌
  Time:2020/05/03

*********************************************/
#include"Thread_Pool.h"
#include<iostream>
int Task::push(Func func) {
 std::unique_lock<std::mutex> lock(mx);
 try {
  tasks.emplace(func);
 }
 catch (std::exception e)
 {
  throw e;
  return -1;
 }
 return 0;
}
int Task::getTaskNum()
{
 return tasks.size();
}
Func Task::pop() {
 std::unique_lock<std::mutex> lock(mx);
 Func temp;
 if (tasks.empty())
  return temp;
 else
 {
  temp = tasks.front();
  tasks.pop();
  return temp;
 }
}
int Thread_Pool::addTasks(Func func)
{
 
 int ret = tasks.push(func);
 cond.notify_one();
 return ret;
}
void Thread_Pool::start() {
 if (!IsStart) {
  IsStart = true;
  for (int i = 0; i < maxThreadNum; i++)
  {
   threads.emplace_back(new std::thread(std::bind(&Thread_Pool::run,this)));   
  }
  
 }
}
void Thread_Pool::run()
{
 while (IsStart)
 {
  Func f;
  if (tasks.getTaskNum() == 0 && IsStart)
  {
   std::unique_lock<std::mutex> lock(tasks.mx);
   cond.wait(lock);
  }
  if (tasks.getTaskNum() != 0 && IsStart)
  {
   f = tasks.pop();
   if(f)
    f();
  }

 }
}
int Thread_Pool::getTaskNum() {
 return tasks.getTaskNum();
}
void Thread_Pool::stop() {

  IsStart = false;
  cond.notify_all();
  for (auto T : threads) {
   std::cout << "線程 " << T->get_id() << " 已停止。" << std::endl;
   T->join();
   if (T != nullptr)
   {
    delete T;
    T = nullptr;
   }
  }
 std::cout << "所有線程已停止。" << std::endl;
}
Thread_Pool::~Thread_Pool() {
 if (IsStart)
 {
  stop();
 }
}

最后是測試用的main.cpp

#include<iostream>
#include"Thread_Pool.h"
using namespace std;
void string_out_one() {
 cout << "One!" << endl;
}
void string_out_two() {
 cout << "Two!" << endl;
}
void string_out_three() {
 cout << "Three!" << endl;
}
int main() {
 {
  Thread_Pool Pool;
  try {
   Pool.start();
  }
  catch (std::exception e)
  {
   throw e;
   cout << "線程池創(chuàng)建失敗。" << endl;
  }
  for (int i = 0; i < 50000 ;)
  {  
   if (Pool.getTaskNum() < 1000) {
    Pool.addTasks(string_out_one);
    Pool.addTasks(string_out_two);
    Pool.addTasks(string_out_three);
    std::cout << i++ << std::endl;
   }
  }
  getchar();
 }
 getchar();
 return 0;
}

執(zhí)行的效果如下：