c++实现简单的线程池_C 语言

这是对pthread线程的一个简单应用

1.      实现了线程池的概念，线程可以重复使用。
2.      对信号量，互斥锁等进行封装，业务处理函数中只需写和业务相关的代码。
3.      移植性好。如果想把这个线程池代码应用到自己的实现中去，只要写自己的业务处理函数和改写工作队列数据的处理方法就可以了。

Sample代码主要包括一个主程序和两个线程实现类
ThreadTest.cpp：主程序
CThreadManager：线程管理Class，线程池的实现类
CThread：线程Class.

主程序实现方法。

1.      实现main函数和一个需要线程处理的业务函数（例子代码中业务函数是一个简单的计算函数Count）。在main函数中创建CThreadManager的实例，产生线程池。这个时候，把业务函数作为函数指针传到CThreadManager里面，最终会被线程调用。
2.      向工作队列中放入业务函数要处理的数据。
3.      设置信号量，唤醒线程。

// 线程要执行的函数
int Count(int nWork)
{
  int nResult = nWork * nWork;
  printf("count result is %d\n",nResult);
  return 0;
}
int main() {
  // 创建线程管理类的实例，把要执行的线程函数和最大线程数传进去
  CThreadManager* pManager = new CThreadManager(Count, 3);
  // 把要进行计算的数放到工作队列中
  pManager->PushWorkQue(5);
  pManager->PushWorkQue(20);
  // 设置信号量，唤醒线程
  pManager->PostSem();
  pManager->PostSem();
  // 等待子线程执行
  sleep(1);
  return 0;
}

CThreadManager实现的方法

1． 把信号量和互斥锁等封装成自己的函数
2． 在new方法里，循环调用CThread的new方法，启动一定数量（可设定）的线程，产生线程池。
3． 这些线程启动后，就会执行CThreadManager中的ManageFuction函数。这个函数是无限循环的，保证了线程在整个程序的生命周期中不销毁。
4． 在循环处理里面，第一行代码就是等待一个信号量，这个信号量是由主程序进行设置的，这个信号信号量如果没有被设置（代表暂时没有需要处理的工作），所有线程都在这里阻塞着。
4.      一旦信号量被设置，根据Linux线程调度机制，在阻塞的线程队列中，其中一个线程被唤醒，可以执行后面的代码。
5.      从工作队列中取出要进行处理的数据（使用互斥锁进行排他）
6.      通过函数指针调用main函数传过来的业务函数，处理数据。
7.      业务函数执行完之后，线程进入下一个循环，等待新的信号量。

class CThreadManager {
  friend void* ManageFuction(void*);
private:
  sem_t m_sem;  // 信号量
  pthread_mutex_t m_mutex; // 互斥锁
  queue<int> m_queWork; // 工作队列
  list<CThread*> m_lstThread; // 线程list
  int (*m_threadFuction)(int); //函数指针，指向main函数传过来的线程执行函数
public:
  CThreadManager(int (*threadFuction)(int), int nMaxThreadCnt);
  virtual ~CThreadManager();
  int WaitSem();
  int PostSem();
  int LockMutex();
  int UnlockMutex();
  void PushWorkQue(int nWork);
  int PopWorkQue();
  int RunThreadFunction(int nWork);
};
// 线程执行函数，它只是个壳子，处理信号量和互斥锁等，
// 最后调用main函数传过来的线程执行函数来实现业务处理
void* ManageFuction(void* argv)
{
  CThreadManager* pManager = (CThreadManager*)argv;
  // 进行无限循环（意味着线程是不销毁的，重复利用）
  while(true)
  {
    // 线程开启后，就在这里阻塞着，直到main函数设置了信号量
    pManager->WaitSem();
    printf("thread wakeup.\n");
    // 从工作队列中取出要处理的数
    pManager->LockMutex();
    int nWork = pManager->PopWorkQue();
    pManager->UnlockMutex();
    printf("call Count function.\n");
    pManager->RunThreadFunction(nWork);
  }
  return 0;
}
// 构造方法
CThreadManager::CThreadManager(int (*threadFuction)(int), int nMaxThreadCnt)
{
  sem_init(&m_sem, 0, 0);
  pthread_mutex_init(&m_mutex, NULL);
  m_threadFuction = threadFuction;
  for(int i=0; i<nMaxThreadCnt; i++)
  {
    CThread* pThread = new CThread(ManageFuction, this);
    printf("thread started.\n");
    m_lstThread.push_back(pThread);
  }
}

CThread实现的方法

CThreadManager比较简单，封装了创建线程和join线程的函数。

CThread::CThread(void* (*threadFuction)(void*),void* threadArgv)
{
  // 初始化线程属性
  pthread_attr_t threadAttr;
  pthread_attr_init(&threadAttr);

  pthread_create(&m_thread, &threadAttr, threadFuction, threadArgv);
}

c++线程池，继承CDoit，实现其中的start和end

/*
 * 多线程管理类
 *
 */

#ifndef CTHREADPOOLMANAGE_H
#define CTHREADPOOLMANAGE_H
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <list>
#include <vector>
#include <time.h>
#include <asm/errno.h>

#define USLEEP_TIME 100
#define CHECK_TIME 1

using namespace std;
class CDoit
{
public:
 virtual int start(void *){};
 virtual int end(){};
};

class CthreadPoolManage
{
private:
 int _minThreads;  //最少保留几个线程
 int _maxThreads;  //最多可以有几个线程
 int _waitSec;      //空闲多少秒后将线程关闭
 class threadInfo{
  public:
  threadInfo(){
   isbusy = false;
   doFlag = true;
  }
  //
  pthread_mutex_t mtx=PTHREAD_MUTEX_INITIALIZER;
  pthread_cond_t cond=PTHREAD_COND_INITIALIZER;
  bool isbusy;   //是否空闲
  bool doFlag;
  //
  time_t beginTime;     //线程不工作开始时间
  pthread_t cThreadPid; //线程id
  pthread_attr_t cThreadAttr; //线程属性
  CDoit * doit;         //任务类
  void  * value;        //需要传递的值
 };
 //线程函数
 static void* startThread(void*);
 //任务队列锁
 pthread_mutex_t _duty_mutex;
 //任务队列
 list<threadInfo*> _dutyList;
 //线程队列锁
 pthread_mutex_t _thread_mutex;
 //线程队列
 list<threadInfo*> _threadList;

///初始化，创建最小个数线程///
void initThread();
///任务分配线程///
static void* taskAllocation(void*arg);
pthread_t tasktPid;
///线程销毁、状态检查线程///
static void* checkThread(void* arg);
pthread_t checktPid;
bool  checkrun;

//线程异常退出清理
static void threadCleanUp(void* arg);

//
int addThread(list<threadInfo*> *plist,threadInfo* ptinfo);

public:
CthreadPoolManage();
/*
保留的最少线程，最多线程数，空闲多久销毁，保留几个线程的冗余
 */
CthreadPoolManage(int min,int max,int waitSec);
~CthreadPoolManage();

int start();
//任务注入器
int putDuty(CDoit *,void *);

int getNowThreadNum();

};

#endif // CTHREADPOOLMANAGE_H

CPP

/*
 * 线程池，线程管理类
 *
 */

#include "cthreadpoolmanage.h"

CthreadPoolManage::CthreadPoolManage()
{
 _minThreads = 5;  //最少保留几个线程
 _maxThreads = 5;  //最多可以有几个线程
 _waitSec = 10;      //空闲多少秒后将线程关闭
 pthread_mutex_init(&_duty_mutex, NULL);
 pthread_mutex_init(&_thread_mutex, NULL);
 checkrun = true;
}

CthreadPoolManage::CthreadPoolManage(int min, int max, int waitSec)
{
  CthreadPoolManage();
  _minThreads = min;  //最少保留几个线程
  _maxThreads = max;  //最多可以有几个线程
  _waitSec = waitSec;      //空闲多少秒后将线程关闭
}

CthreadPoolManage::~CthreadPoolManage()
{

}
void CthreadPoolManage::threadCleanUp(void* arg)
{
 threadInfo* tinfo = (threadInfo*)arg;
 tinfo->isbusy = false;
 pthread_mutex_unlock(&tinfo->mtx);
 pthread_attr_destroy (&tinfo->cThreadAttr);
 delete tinfo;
}

void* CthreadPoolManage::startThread(void* arg)
{
 cout<<"线程开始工作"<<endl;
 threadInfo* tinfo = (threadInfo*)arg;
 pthread_cleanup_push(threadCleanUp,arg);
 while(tinfo->doFlag){
  pthread_mutex_lock(&tinfo->mtx);
  if(tinfo->doit == NULL)
  {
   cout<<"开始等待任务"<<endl;
   pthread_cond_wait(&tinfo->cond,&tinfo->mtx);
   cout<<"有任务了"<<endl;
  }
  tinfo->isbusy = true;
  tinfo->doit->start(tinfo->value);
  tinfo->doit->end();
  tinfo->doit=NULL;
  tinfo->isbusy = false;
  time( &tinfo->beginTime);
  pthread_mutex_unlock(&tinfo->mtx);
 }
 //0正常执行到这儿不执行清理函数，异常会执行
 pthread_cleanup_pop(0);
 pthread_attr_destroy (&tinfo->cThreadAttr);
 delete tinfo;
 cout<<"线程结束"<<endl;
}

void CthreadPoolManage::initThread()
{
 int i = 0;
 for(i = 0;i<this->_minThreads;i++)
 {
   threadInfo *tinfo = new threadInfo;
   tinfo->doit = NULL;
   tinfo->value = NULL;
   tinfo->isbusy = false;
   tinfo->doFlag = true;
  // PTHREAD_CREATE_DETACHED (分离线程) 和 PTHREAD _CREATE_JOINABLE (非分离线程)
   pthread_attr_init(&tinfo->cThreadAttr);
   pthread_attr_setdetachstate(&tinfo->cThreadAttr,PTHREAD_CREATE_DETACHED );
   cout<<"初始化了一个线程"<<endl;
   if(pthread_create(&tinfo->cThreadPid,&tinfo->cThreadAttr,startThread,(void *)tinfo) != 0)
  {
  cout<<"创建线程失败"<<endl;
  break;
  }
  this->_threadList.push_back(tinfo);
 }
}

int CthreadPoolManage::addThread(std::list< CthreadPoolManage::threadInfo* >* plist, CthreadPoolManage::threadInfo* ptinfo)
{
   threadInfo *tinfo = new threadInfo;
   tinfo->doit = ptinfo->doit;
   tinfo->value = ptinfo->value;
   tinfo->isbusy = true;
   if(pthread_create(&tinfo->cThreadPid,NULL,startThread,(void *)tinfo) != 0)
  {
  cout<<"创建线程失败"<<endl;
  return -1;
  }
  plist->push_back(tinfo);
  return 0;
}

int CthreadPoolManage::putDuty(CDoit* doit, void* value)
{
 threadInfo *tinfo = new threadInfo;
 time( &tinfo->beginTime);
 tinfo->doit= doit;
 tinfo->value = value;
 pthread_mutex_lock(&_duty_mutex);
  this->_dutyList.push_back(tinfo);
 pthread_mutex_unlock(&_duty_mutex);
 return 0;
}

void* CthreadPoolManage::taskAllocation(void*arg)
{
 CthreadPoolManage * ptmanage = (CthreadPoolManage*)arg;
 int size_1 = 0;
 int size_2 = 0;
 int i_1 = 0;
 int i_2 = 0;
 bool a_1 = true;
 bool a_2 = true;
 threadInfo* ptinfo;
 threadInfo* ptinfoTmp;
 while(true){
   size_1 = 0;
   size_2 = 0;
   pthread_mutex_lock(&ptmanage->_duty_mutex);
   pthread_mutex_lock(&ptmanage->_thread_mutex);
   size_1 = ptmanage->_dutyList.size();
   size_2 =ptmanage->_threadList.size();
   for(list<threadInfo*>::iterator itorti1 = ptmanage->_dutyList.begin();itorti1 !=ptmanage->_dutyList.end();)
   {
  ptinfo = *itorti1;
  a_1 = true;
  for(list<threadInfo*>::iterator itorti2 = ptmanage->_threadList.begin();itorti2!=ptmanage->_threadList.end();itorti2++){
   ptinfoTmp = *itorti2;
   if(EBUSY == pthread_mutex_trylock(&ptinfoTmp->mtx))
   {
    continue;
   }
   if(!ptinfoTmp->isbusy)
   {
    ptinfoTmp->doit = ptinfo->doit;
    ptinfoTmp->value = ptinfo->value;
    ptinfoTmp->isbusy = true;
    pthread_cond_signal(&ptinfoTmp->cond);
    pthread_mutex_unlock(&ptinfoTmp->mtx);
    a_1 = false;
    delete ptinfo;
    break;
   }
   pthread_mutex_unlock(&ptinfoTmp->mtx);
    }
    if(a_1){
   if(ptmanage->_threadList.size()>ptmanage->_maxThreads||ptmanage->addThread(&ptmanage->_threadList,ptinfo)!=0)
   {
    itorti1++;
    continue;
   }else{
    itorti1 = ptmanage->_dutyList.erase(itorti1);
   }
   delete ptinfo;
    }else{
   itorti1 = ptmanage->_dutyList.erase(itorti1);
    }
   }
   pthread_mutex_unlock(&ptmanage->_duty_mutex);
   pthread_mutex_unlock(&ptmanage->_thread_mutex);
   usleep(USLEEP_TIME);
 }
 return 0;
}

void* CthreadPoolManage::checkThread(void* arg)
{
 CthreadPoolManage * ptmanage = (CthreadPoolManage*)arg;
 threadInfo* ptinfo;
 time_t nowtime;
 while(ptmanage->checkrun){
  sleep(CHECK_TIME);
  pthread_mutex_lock(&ptmanage->_thread_mutex);
  if(ptmanage->_threadList.size()<=ptmanage->_minThreads)
  {
   pthread_mutex_unlock(&ptmanage->_thread_mutex);
   continue;
  }
  for(list<threadInfo*>::iterator itorti2 = ptmanage->_threadList.begin();itorti2!=ptmanage->_threadList.end();){
   ptinfo = *itorti2;
   if(EBUSY == pthread_mutex_trylock(&ptinfo->mtx))
  {
   itorti2++;
   continue;
  }
  time(&nowtime);
  if(ptinfo->isbusy == false && nowtime-ptinfo->beginTime>ptmanage->_waitSec)
  {
   ptinfo->doFlag = false;
   itorti2 = ptmanage->_threadList.erase(itorti2);
  }else{
   itorti2++;
  }
  pthread_mutex_unlock(&ptinfo->mtx);
  }
  pthread_mutex_unlock(&ptmanage->_thread_mutex);
 }
}

int CthreadPoolManage::start()
{
 //初始化
 this->initThread();
 //启动任务分配线程
  if(pthread_create(&tasktPid,NULL,taskAllocation,(void *)this) != 0)
  {
  cout<<"创建任务分配线程失败"<<endl;
  return -1;
  }
  //创建现程状态分配管理线程
  if(pthread_create(&checktPid,NULL,checkThread,(void *)this) != 0)
  {
  cout<<"创建线程状态分配管理线程失败"<<endl;
  return -1;
  }
 return 0;
}

///////////////////////////////
int CthreadPoolManage::getNowThreadNum()
{
 int num = 0;
 pthread_mutex_lock(&this->_thread_mutex);
  num = this->_threadList.size();
 pthread_mutex_unlock(&this->_thread_mutex);
 return num ;
}

以上是小编为您精心准备的的内容，在的博客、问答、公众号、人物、课程等栏目也有的相关内容，欢迎继续使用右上角搜索按钮进行搜索c++
线程池
c语言实现线程池、c语言实现简单数据库、链表c语言实现 简单、java线程池简单实现、c语言线程池，以便于您获取更多的相关知识。

时间： 2024-09-30 11:55:42

c++实现简单的线程池_C 语言的相关文章

C语言实现支持动态拓展和销毁的线程池_C 语言

本文实例介绍了C 语言实现线程池,支持动态拓展和销毁,分享给大家供大家参考,具体内容如下 实现功能 1.初始化指定个数的线程 2.使用链表来管理任务队列 3.支持拓展动态线程 4.如果闲置线程过多,动态销毁部分线程 #include <stdio.h> #include <pthread.h> #include <stdlib.h> #include <signal.h> /*线程的任务队列由,函数和参数组成,任务由链表来进行管理*/ typedef str

一个简单的线程池

最近自己,很烦所以超级久没学习了,今天趁着抗战七十周年放三天假,赶紧看下书. 废话不多说. 今天,介绍一个简单的线程池. 首先说明什么是线程池,线程池:是包含若干个线程,来处理多个任务的线程集合. 它的目的是用来处理,大量的相对短暂的任务. 这里我们先来解释下两个概念,什么叫大量呢?对于线程来说,需要线程数小于任务数,第二,短暂的任务是指,任务需要相对短暂,如果线程和主进程同周期,则不适合用线程池. 然后来说下CPU数和线程数的关系,如果你的任务主要是计算密集型任务 则:线程个数 = CPU个数

C实现一个简单的线程池

//threadpool.h #ifndef __THREADPOOL_H__ #define __THREADPOOL_H__ #include <pthread.h> typedef void* (*task_fun)(void*); //用链表来维护等待任务 typedef struct threadtask { //任务的执行函数 task_fun task; //执行函数的参数 void* arg; //下一节点 struct threadtask* next; }THREAD_TA

实例分析一个简单的Win32程序_C 语言

本文较为详细的分析了一个Win32程序的组成.结构.实现方法及运行原理,对于进行Windows程序设计有很好的借鉴参考价值.分享给大家供大家参考之用.具体分析如下: 一.Windows程序与普通C或C++程序的不同 学过C或C++等语言的人都知道,我们写的程序都一个入口,main函数,但是在Win32程序里,我们的入口函数又是什么呢?它是怎么样运行的,跟我们用C或C++写的控制台程序又有什么不同呢? 我们先说Win32程序跟我们控制台的程序的一个很重要的不同点,就是Win32程序是一个消息响应程

用Python实现一个简单的线程池_python

线程池的概念是什么? 在面向对象编程中,创建和销毁对象是很费时间的,因为创建一个对象要获取内存资源或者其它更多资源.在Java中更是 如此,虚拟机将试图跟踪每一个对象,以便能够在对象销毁后进行垃圾回收.所以提高服务程序效率的一个手段就是尽可能减少创建和销毁对象的次数,特别是一些 很耗资源的对象创建和销毁.如何利用已有对象来服务就是一个需要解决的关键问题,其实这就是一些"池化资源"技术产生的原因. 我理解为线程池是一个存放很多线程的单位,同时还有一个对应的任务队列.整个执行过程其实就是使

C++编写简单的打靶游戏_C 语言

首次自己写程序,很不完善,还有许多问题需要解决...见谅见谅 #define GDIPVER 0x0110 #define WIN32_LEAN_AND_MEAN #include <windows.h> #include <ObjIdl.h> #include <GdiPlus.h> #include <windowsx.h> #include <tchar.h> #include <mmsystem.h> #include <

C++实现简单的图书管理系统_C 语言

今天再为大家介绍另一个常用的管理系统--图书管理系统,希望大家可以亲自动手实践一下,下面就与大家一起分享我的劳动成果. 图书信息包括:登录号.书名.作者名.分类号.出版单位.出版时间.价格等.试设计一图书信息管理系统,使之能提供以下功能: (1)图书信息录入功能(图书信息用文件保存) (2)图书信息浏览功能 (3)查询和排序功能:(至少一种查询方式)         .按书名查询         .按作者名查询 (4)图书信息的删除与修改 分享代码如下 #include<iostream.h>

C++设计模式之简单工厂模式实例_C 语言

问题描述 之前在公司做了一个windows 8平台的阅读器.首先,需要将电子书中的内容渲染到屏幕上,而电子书每一页都包含各种各样的内容,比如:图形.图像和文字等等:不同的内容,就是不同的对象:在将不同的内容渲染到屏幕上之前,就需要new操作,建立不同的对象,然后再在屏幕上进行描绘.这个时候,就需要进行很多new操作,new操作分布在代码的不同地方,管理起来很麻烦,而且也很乱,到后期扩展和维护的时候,有的时候,对象多的让开发人员不知道这个对象是干什么的,这就增加了难度:同时,new操作,都会有对应

c++编写简单的计算器程序_C 语言

首先来看下本人的开发环境 系统:win7 电脑:dell 运行环境:vs2015 语言:c++ 简单计算器代码 //四则运算 #include "stdafx.h" #include<iostream> #include<stdio.h> using namespace std; void add() { printf("输入要计算的加数(例如a b)\n"); int adda=0, addb=0,addc=0; cin >>