threadpool.cpp
#include<iostream> #include<unistd.h> #include<sys/types.h> #include<sys/wait.h> #include<queue> class Start{ public: Start(int id) :id_(id) {} virtual void Run(){ printf("Start is running!!!\n"); } int id_; }; class ThreadPool{ public: ThreadPool(int threadSize) :threadSize_(threadSize) ,threadCurNum_(0) { pthread_mutex_init(&mutex_,NULL); pthread_cond_init(&cond_,NULL); isQuit_=false; pthread_t tid; for(int i=0;i<threadSize;++i){ pthread_create(&tid,NULL,threadStart,this); ++threadCurNum_; } } ~ThreadPool(){ pthread_mutex_destroy(&mutex_); pthread_cond_destroy(&cond_); } void push(Start* data){ pthread_mutex_lock(&mutex_); if(isQuit_){ pthread_mutex_unlock(&mutex_); //释放线程就不允许再增加数据了 return; } que_.push(data); pthread_mutex_unlock(&mutex_); pthread_cond_signal(&cond_); } void pop(Start** data){ pthread_mutex_lock(&mutex_); while(0==que_.size()){ if(isQuit_){ //释放线程池 --threadCurNum_; pthread_mutex_unlock(&mutex_); pthread_exit(NULL); //线程退出 } pthread_cond_wait(&cond_,&mutex_); } *data=que_.front(); que_.pop(); pthread_mutex_unlock(&mutex_); } void threadClear(){ pthread_mutex_lock(&mutex_); isQuit_=true; //标记要释放线程池里的线程 pthread_mutex_unlock(&mutex_); if(threadCurNum_>0) pthread_cond_broadcast(&cond_); //唤醒所有线程去争抢锁资源 } private: static void *threadStart(void *arg){ ThreadPool *ptr=(ThreadPool*)arg; while(1){ Start *p=nullptr; ptr->pop(&p); p->Run(); delete[] p; } } int threadSize_; std::queue<Start*> que_; bool isQuit_; int threadCurNum_; pthread_mutex_t mutex_; pthread_cond_t cond_; };main.cpp
#include "threadpool.cpp" using namespace std; #define ThreadSize 10 #define MyStartCount 100 class MyStart:public Start{ public: MyStart(int id) :Start(id) {} void Run()override{ cout<<"id= "<<id_<<",thread="<<pthread_self()<<endl; usleep(300); } }; int main(){ ThreadPool pool(ThreadSize); for(int i=0;i<MyStartCount;++i){ MyStart *p=new MyStart(i); pool.push(p); } //while(1) sleep(6); pool.threadClear(); return 0; }blockingqueue.hpp
#pragma once #include<iostream> #include<semaphore.h> #include<vector> using namespace std; template<class T> class BlockingQueue { public: BlockingQueue(int maxsize=100)//默认阻塞队列大小为100 :_queue(maxsize) ,_head(0) ,_tail(0) ,_size(0) ,_maxsize(maxsize) { sem_init(&_lock,0,1); sem_init(&_used,0,0); sem_init(&_rem,0,maxsize); } ~BlockingQueue() { sem_destroy(&_lock); sem_destroy(&_used); sem_destroy(&_rem); } void Push(const T& data) { sem_wait(&_rem); sem_wait(&_lock); _queue[_tail]=data; _tail=(_tail+1)%_maxsize; ++_size; sem_post(&_lock); sem_post(&_used); } void Pop(T& data) { sem_wait(&_used); sem_wait(&_lock); data=_queue[_head]; _head=(_head+1)%_maxsize; --_size; sem_post(&_lock); sem_post(&_rem); } private: int _head; int _tail; int _size; int _maxsize; vector<T> _queue; sem_t _lock;//互斥锁 sem_t _used;//已使用 sem_t _rem;//未使用 };threadpool.hpp
#pragma once #include"blockingqueue.hpp" class Stack { public: virtual void Run() { cout<<"Stack-----\n"; } virtual ~Stack() {} }; //线程池 class ThreadPool { public: ThreadPool(int thread_size,int queue_size=100)//默认阻塞队列大小为100 :_queue(queue_size) ,_size(thread_size) { for(int i=0;i<_size;++i) { pthread_t tid; pthread_create(&tid,NULL,PthreadEntry,this);//调用函数必须为静态且需要传入this指针 tid_arr.push_back(tid); } } ~ThreadPool() { for(int i=0;i<_size;++i) { pthread_cancel(tid_arr[i]); } for(int i=0;i<_size;++i) { pthread_join(tid_arr[i],NULL); } } void AddStack(Stack* ptr)//用父类指针接收，多态的特性 { _queue.Push(ptr); } private: BlockingQueue<Stack*> _queue;//阻塞队列里存放线程执行代码的父类指针，使用多态调用 int _size;//线程个数 vector<pthread_t> tid_arr;//线程tid数组 static void* PthreadEntry(void* arg) { ThreadPool* ptr=(ThreadPool*)arg;//接收this指针 while(1) { Stack* pst=NULL; ptr->_queue.Pop(pst);//从阻塞队列中取执行代码 pst->Run();//多态 delete pst; } } };main.cc
#include"threadpool.hpp" #include<sys/syscall.h> #include<unistd.h> #include<stdio.h> class MyStack:public Stack{ public: MyStack(int n) :_id(n) {} void Run()override//重写父类的函数，改为自己需要执行的代码 { printf("MyStack---tid=%d,id=%d\n",syscall(SYS_gettid),_id); } ~MyStack()override//重写析构函数，释放子类的资源 {} private: int _id; }; int main() { ThreadPool pool(10);//创建10个进程 for(int i=0;i<20;++i) { pool.AddStack(new MyStack(i)); } while(1) sleep(1); return 0; }