网上分享的基本上是属于纯模板的。不利于新手学习,之前在看视频写项目的时候老师也是用的这个模板。虽然很好,用到的都是可变参数,匿名表达式,和其他一些函数。后来自己到github上找到了一个C++版本的。稍作修改。弄了一个模板。算是比较简洁的
线程池需要的:
1.线程池的h和cpp文件(线程池创建,运行,添加任务,唤醒线程工作,销毁)
2.任务的h和cpp文件,(任务类型的创建,任务需要做的事),这里为了简单。我创建的任务就是简单打印线程执行的信息。具体任务可以自己定义。我在自己的http服务器上就定义的是epoll创建监听连接的任务
3.锁的h和cpp文件。用了互斥锁(mutex)和条件变量(condition_variable)来保证消息队列的任务只能一个线程拿到并且执行
ThreadPoll.h
#pragma once
#include "locker.h"
#include <queue>
using namespace std;
template< typename T >
class ThreadPool {
private:
int thread_number; // 线程池的线程数
pthread_t *threads; // 线程数组
queue<T *> task_queue; // 任务队列
MutexLocker queue_mutex_locker; // 任务队列的互斥锁
Cond queue_cond_locker; // 任务队列的条件变量
bool m_stop; // 是否结束线程
public:
ThreadPool( int thread_num = 20 );
~ThreadPool();
bool append( T *task ); // 向任务队列添加任务
private:
static void *worker( void * ); // 工作线程
void run(); // 线程池中线程开始运行的函数
T *getTask(); // 从任务队列中获取队首的任务
};
ThreadPool.cpp
#Include"ThreadPool.h"
template< typename T >
ThreadPool<T>::ThreadPool( int thread_num ) :thread_number(thread_num),
threads(NULL), m_stop(false) {
if( thread_number < 0 ) {
cout << "thread_number < 0\n";
throw exception();
}
// 创建数组存放线程号
threads = new pthread_t[ thread_number ];
if( !threads ) {
cout << "threads is NULL\n";
throw exception();
}
// 创建规定数量的线程
for( int i = 0; i < thread_number; i++ ) {
// 由于pthread_create第三个参数必须指向静态函数,要使用类成员函数和变量,只能通过:
// 1) 类的静态对象
// 2) 将类的对象作为参数传给静态函数
// 这里通过第二种方法实现
if( pthread_create( &threads[i], NULL, worker, this ) ) { // 成功返回0
delete[] threads; // 创建失败则释放所有已分配的空间
cout << "pthread_create error\n";
throw exception();
}
// 将线程进行脱离,线程运行完后自动回收,避免使用主线程进行join等待其结束
if( pthread_detach( threads[i] ) ) {
delete[] threads;
cout << "pthread_detach error\n";
throw exception();
}
}
}
// 析构函数中,将m_stop置true,此时将阻塞中的所有线程唤醒
// 由于 !m_stop 为false,线程会退出循环,线程结束被回收( 详见函数run() )
// 若不唤醒线程,则在程序退出后,线程非正常结束,同时会导致
template< typename T >
ThreadPool<T>::~ThreadPool() {
delete[] threads;
m_stop = true;
queue_cond_locker.broadcast();
}
/* 添加任务时需要先上锁,并判断队列是否为空 */
template< typename T >
bool ThreadPool<T>::append( T *task ) {
queue_mutex_locker.mutex_lock();
bool need_signal = task_queue.empty(); // 记录添加任务之前队列是否为空
task_queue.push( task );
queue_mutex_locker.mutex_unlock();
// 如果添加任务之前队列为空,即所有线程都在wait,所以需要唤醒某个线程
if( need_signal ) {
queue_cond_locker.signal();
}
return true;
}
//让线程活起来,准备接收工作
template< typename T >
void * ThreadPool<T>::worker( void *arg ) {
ThreadPool *pool = ( ThreadPool * )arg;
pool->run();
return pool;
}
// 获取处于队首的任务,获取时需要加锁,避免发生错误
// 若队列为空,则返回NULL,该线程成为等待状态(详见函数run())
template< typename T >
T* ThreadPool<T>::getTask() {
T *task = NULL;
queue_mutex_locker.mutex_lock();
if( !task_queue.empty() ) {
task = task_queue.front();
task_queue.pop();
}
queue_mutex_locker.mutex_unlock();
return task;
}
template< typename T >
void ThreadPool<T>::run() {
while( !m_stop ) { // 当线程池没有结束时,线程循环获取任务进行执行
T *task = getTask();
if( !task ) {
queue_cond_locker.wait(); // 队列为空,线程开始等待
} else {
task->doit(); // 开始执行任务
delete task; //task指向的对象在WebServer中new出来,因此需要手动delete
}
}
}
Task.h
#pragma once
#include<iostream>
#include<thread>
class Task
{
public:
Task();
~Task();
void doit();
};
Task::Task()
{
}
Task::~Task()
{
}
inline void Task::doit()
{
std::this_thread::sleep_for(std::chrono::milliseconds(50));
//任务就是休眠0.05秒然后打印线程id
std::cout << "worker thread ID:" << std::this_thread::get_id() << std::endl;
}
Lock.h
#pragma once
#include <iostream>
#include <exception>
#include <pthread.h>
using namespace std;
/* 线程锁 */
class MutexLocker {
private:
pthread_mutex_t m_mutex;
public:
MutexLocker() { //初始化
if (pthread_mutex_init(&m_mutex, NULL)) {
cout << "mutex init errror __ 1\n";
throw exception();
}
}
~MutexLocker() {
pthread_mutex_destroy(&m_mutex);
}
bool mutex_lock() {
return pthread_mutex_lock(&m_mutex) == 0;
}
bool mutex_unlock() {
return pthread_mutex_unlock(&m_mutex);
}
};
/* 条件变量 */
class Cond {
private:
pthread_mutex_t m_mutex;
pthread_cond_t m_cond;
public:
Cond() {
if (pthread_mutex_init(&m_mutex, NULL)) {
throw exception();
}
if (pthread_cond_init(&m_cond, NULL)) {
pthread_cond_destroy(&m_cond);
throw exception();
}
}
~Cond() {
pthread_mutex_destroy(&m_mutex);
pthread_cond_destroy(&m_cond);
}
// 等待条件变量,cond与mutex搭配使用,避免造成共享数据的混乱
bool wait() {
pthread_mutex_lock(&m_mutex);
int ret = pthread_cond_wait(&m_cond, &m_mutex);
pthread_mutex_unlock(&m_mutex);
return ret == 0;
}
// 唤醒等待该条件变量的某个线程
bool signal() {
return pthread_cond_signal(&m_cond) == 0;
}
// 唤醒所有等待该条件变量的线程
bool broadcast() {
return pthread_cond_broadcast(&m_cond) == 0;
}
};
main.cpp
#include <cstdio>
#include"threads.h"
#include"Task.h"
int main()
{
std::cout << "main thread:" << pthread_self() << std::endl;
int n = 100;
ThreadPool<Task> pool(5);
Task** arr= new Task*[n];
for (int i = 0; i < n; i++)
{
arr[i] = new Task(i);//创建n个任务
}
for (int i = 0; i < n; i++)
{
pool.append(arr[i]);//添加这些任务
}
std::this_thread::sleep_for(std::chrono::milliseconds(10000));
std::cout << "main ID:" << std::this_thread::get_id() << std::endl;
//删除任务在线程执行完后,线程那里会删除释放这个任务,无需自己释放
//详情见ThreadPool.cpp的run()函数delete task;
return 0;
}
运行截图如下: