Chinaunix首页 | 论坛 | 博客
  • 博客访问: 4244544
  • 博文数量: 776
  • 博客积分: 13014
  • 博客等级: 上将
  • 技术积分: 10391
  • 用 户 组: 普通用户
  • 注册时间: 2010-02-22 17:00
文章分类

全部博文(776)

文章存档

2015年(55)

2014年(43)

2013年(147)

2012年(20)

2011年(82)

2010年(429)

分类: LINUX

2010-11-06 00:24:22

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

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

1 线程的创建和取消函数

pthread_create -- 创建线程

pthread_join -- 合并线程

pthread_cancel -- 取消线程

2 线程同步函数

pthread_mutex_lock

pthread_mutex_unlock

pthread_cond_signal

pthread_cond_wait

线程池的实现:

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

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

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

C代码

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

线程池类

/**
线程池
**/

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

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

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

AddTask函数把任务添加到线程池的任务列表中,并通知线程进行处理。当任务到到时,把任务放入m_vecTaskList任务列表中,并用pthread_cond_signal唤醒一个线程进行处理。

StopAll函数停止所有的线程

 

Cpp代码
 
************************************************  
 
代码:  
 
××××××××××××××××××××CThread.h  
 
   
 
#ifndef __CTHREAD  
#define __CTHREAD  
#include   
#include   
#include   
 
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 m_vecTaskList;         //任务列表  
        int m_iThreadNum;                            //线程池中启动的线程数             
        static vector m_vecIdleThread;   //当前空闲的线程集合  
        static vector 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
#include
using namespace std;
void CTask::SetData(void * data)
{
        m_ptrData = data;
}
vector CThreadPool::m_vecBusyThread;
vector 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::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::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* taskList = (vector*)threadData;
                vector::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::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
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;
}
阅读(1181) | 评论(0) | 转发(1) |
给主人留下些什么吧!~~