主题：一个简单的linux线程池

怼烎@ 2022-07-12 07:53 159阅读 0赞

**线程池** ：简单地说，线程池 就是预先创建好一批线程，方便、快速地处理收到的业务。比起传统的到来一个任务，即时创建一个线程来处理，节省了线程的创建和回收的开销，响应更快，效率 更高。

在linux中，使用的是posix线程库，首先介绍几个常用的函数：

1 线程的创建和取消函数

pthread\_create

创建

pthread\_join

合并线程

pthread\_cancel

取消线程

2 线程同步函数

pthread\_mutex\_lock

pthread\_mutex\_unlock

pthread\_cond\_signal

pthread\_cond\_wait

关于函数的详细说明，参考man手册

**线程池的实现：**

线程池的实现主要分为三部分，线程的创建、添加任务到线程池中、工作线程从任务队列中取出任务进行处理。

主要有两个类来实现，CTask,CThreadPool

/\*\*  
执行任务的类，设置任务数据并执行  
\*\*/

class CTask \{ protected: string m\_strTaskName; //任务的名称 void\* m\_ptrData; //要执行的任务的具体数据 public: CTask()\{\} CTask(string taskName) \{ this->m\_strTaskName = taskName; m\_ptrData = NULL; \} virtual int Run()= 0; void SetData(void\* data); //设置任务数据 \};

任务类是个虚类，所有的任务要从CTask类中继承 ，实现run接口，run接口中需要实现的就是具体解析任务的逻辑。m\_ptrData是指向任务数据的指针，可以是简单数据类型，也可以是自定义的复杂 数据类型。

线程池类

/\*\*  
线程池  
\*\*/

class CThreadPool \{ private: vector<CTask\*> m\_vecTaskList; //任务列表 int m\_iThreadNum; //线程池中启动的线程数 static vector<pthread\_t> m\_vecIdleThread; //当前空闲的线程集合 static vector<pthread\_t> m\_vecBusyThread; //当前正在执行的线程集合 static pthread\_mutex\_t m\_pthreadMutex; //线程同步锁 static pthread\_cond\_t m\_pthreadCond; //线程同步的条件变量 protected: static void\* ThreadFunc(void \* threadData); //新线程的线程函数 static int MoveToIdle(pthread\_t tid); //线程执行结束后，把自己放入到空闲线程中 static int MoveToBusy(pthread\_t tid); //移入到忙碌线程中去 int Create(); //创建所有的线程 public: CThreadPool(int threadNum); int AddTask(CTask \*task); //把任务添加到线程池中 int StopAll(); \};

当线程池对象创建后，启动一批线程，并把所有的线程放入空闲列表中，当有任务到达时，某一个线程取出任务并进行处理。

线程之间的同步用线程锁和条件变量。

这个类的对外接口有两个：

AddTask函数把任务添加到线程池的任务列表中，并通知线程进行处理。当任务到到时，把任务放入m\_vecTaskList任务列表中，并用 pthread\_cond\_signal唤醒一个线程进行处理。

StopAll函数停止所有的线程

\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* 代码: ××××××××××××××××××××CThread.h \#ifndef \_\_CTHREAD \#define \_\_CTHREAD \#include <vector> \#include <string> \#include <pthread.h> using namespace std; /\*\* 执行任务的类，设置任务数据并执行 \*\*/ class CTask \{ protected: string m\_strTaskName; //任务的名称 void\* m\_ptrData; //要执行的任务的具体数据 public: CTask()\{\} CTask(string taskName) \{ this->m\_strTaskName = taskName; m\_ptrData = NULL; \} virtual int Run()= 0; void SetData(void\* data); //设置任务数据 \}; /\*\* 线程池 \*\*/ class CThreadPool \{ private: vector<CTask\*> m\_vecTaskList; //任务列表 int m\_iThreadNum; //线程池中启动的线程数 static vector<pthread\_t> m\_vecIdleThread; //当前空闲的线程集合 static vector<pthread\_t> m\_vecBusyThread; //当前正在执行的线程集合 static pthread\_mutex\_t m\_pthreadMutex; //线程同步锁 static pthread\_cond\_t m\_pthreadCond; //线程同步的条件变量 protected: static void\* ThreadFunc(void \* threadData); //新线程的线程函数 static int MoveToIdle(pthread\_t tid); //线程执行结束后，把自己放入到空闲线程中 static int MoveToBusy(pthread\_t tid); //移入到忙碌线程中去 int Create(); //创建所有的线程 public: CThreadPool(int threadNum); int AddTask(CTask \*task); //把任务添加到线程池中 int StopAll(); \}; \#endif 类的实现为： ××××××××××××××××××××CThread.cpp \#include "CThread.h" \#include <string> \#include <iostream> using namespace std; void CTask::SetData(void \* data) \{ m\_ptrData = data; \} vector<pthread\_t> CThreadPool::m\_vecBusyThread; vector<pthread\_t> CThreadPool::m\_vecIdleThread; pthread\_mutex\_t CThreadPool::m\_pthreadMutex = PTHREAD\_MUTEX\_INITIALIZER; pthread\_cond\_t CThreadPool::m\_pthreadCond = PTHREAD\_COND\_INITIALIZER; CThreadPool::CThreadPool(int threadNum) \{ this->m\_iThreadNum = threadNum; Create(); \} int CThreadPool::MoveToIdle(pthread\_t tid) \{ vector<pthread\_t>::iterator busyIter = m\_vecBusyThread.begin(); while(busyIter != m\_vecBusyThread.end()) \{ if(tid == \*busyIter) \{ break; \} busyIter++; \} m\_vecBusyThread.erase(busyIter); m\_vecIdleThread.push\_back(tid); return 0; \} int CThreadPool::MoveToBusy(pthread\_t tid) \{ vector<pthread\_t>::iterator idleIter = m\_vecIdleThread.begin(); while(idleIter != m\_vecIdleThread.end()) \{ if(tid == \*idleIter) \{ break; \} idleIter++; \} m\_vecIdleThread.erase(idleIter); m\_vecBusyThread.push\_back(tid); return 0; \} void\* CThreadPool::ThreadFunc(void \* threadData) \{ pthread\_t tid = pthread\_self(); while(1) \{ pthread\_mutex\_lock(&m\_pthreadMutex); pthread\_cond\_wait(&m\_pthreadCond,&m\_pthreadMutex); cout << "tid:" << tid << " run" << endl; //get task vector<CTask\*>\* taskList = (vector<CTask\*>\*)threadData; vector<CTask\*>::iterator iter = taskList->begin(); while(iter != taskList->end()) \{ MoveToBusy(tid); break; \} CTask\* task = \*iter; taskList->erase(iter); pthread\_mutex\_unlock(&m\_pthreadMutex); cout << "idel thread number:" << CThreadPool::m\_vecIdleThread.size() << endl; cout << "busy thread number:" << CThreadPool::m\_vecBusyThread.size() << endl; //cout << "task to be run:" << taskList->size() << endl; task->Run(); //cout << "CThread::thread work" << endl; cout << "tid:" << tid << " idle" << endl; \} return (void\*)0; \} int CThreadPool::AddTask(CTask \*task) \{ this->m\_vecTaskList.push\_back(task); pthread\_cond\_signal(&m\_pthreadCond); return 0; \} int CThreadPool::Create() \{ for(int i = 0; i < m\_iThreadNum;i++) \{ pthread\_t tid = 0; pthread\_create(&tid,NULL,ThreadFunc,&m\_vecTaskList); m\_vecIdleThread.push\_back(tid); \} return 0; \} int CThreadPool::StopAll() \{ vector<pthread\_t>::iterator iter = m\_vecIdleThread.begin(); while(iter != m\_vecIdleThread.end()) \{ pthread\_cancel(\*iter); pthread\_join(\*iter,NULL); iter++; \} iter = m\_vecBusyThread.begin(); while(iter != m\_vecBusyThread.end()) \{ pthread\_cancel(\*iter); pthread\_join(\*iter,NULL); iter++; \} return 0; \} 简单示例： ××××××××test.cpp \#include "CThread.h" \#include <iostream> using namespace std; class CWorkTask: public CTask \{ public: CWorkTask() \{\} int Run() \{ cout << (char\*)this->m\_ptrData << endl; sleep(10); return 0; \} \}; int main() \{ CWorkTask taskObj; char szTmp\[\] = "this is the first thread running,haha success"; taskObj.SetData((void\*)szTmp); CThreadPool threadPool(10); for(int i = 0;i < 11;i++) \{ threadPool.AddTask(&taskObj); \} while(1) \{ sleep(120); \} return 0; \}

转载自：http://www.javaeye.com/topic/345688