【Linux】简单的线程池

线程池介绍

基本概念

定义

组成部分

线程池的优点

资源高效

响应迅速

可管理性

线程池的工作原理

线程池的使用场景

线程池的注意事项

实现简单的线程池

前置函数

Mutex 类介绍

LockGuard 类介绍

Log类的介绍

枚举定义

Log类

全局对象

Conf类

myThread类的介绍

类成员变量

构造函数

成员函数

使用场景

注意事项

Task类的介绍

成员变量

成员函数

错误处理

使用场景

注意事项

线程池的实现

ThreadPool 类

成员变量

成员函数

线程池介绍

基本概念

线程池是一种多线程处理形式，它用于高效地处理大量并发任务，通过重用已创建的线程来避免频繁地线程创建与销毁所带来的开销。

定义

线程池：一种维护多个线程并等待执行任务的系统。
任务：任何可以执行的代码片段，如函数或方法。

组成部分

任务队列：存放待执行的任务。
线程集合：一组可并行执行任务的线程。
线程池管理器：负责线程的创建、销毁、管理，以及任务的调度。

线程池的优点

资源高效

通过重用线程，降低了线程创建和销毁的频率，从而节省了系统资源。

响应迅速

预先创建的线程处于等待状态，任务到达时无需等待，可立即执行。

可管理性

提供了统一的管理和监控接口，便于跟踪线程状态和行为。

线程池的工作原理

任务提交：当有新任务时，将其提交到线程池。
线程分配：线程池管理器检查是否有空闲线程，若有，则分配任务；若无，则任务排队等待。
任务执行与调度：线程完成任务后，从队列中取出下一个任务继续执行。

线程池的使用场景

适用于处理大量并发任务，特别是任务执行时间较短的情况。
常用于服务器程序，如Web服务器和数据库服务器，以应对大量并发请求。

线程池的注意事项

合理设置线程池大小：避免线程过多导致的资源竞争和性能下降。
监控线程池状态：确保任务的正常执行，并及时发现潜在问题。
同步与互斥：注意数据的一致性和线程安全性，避免数据竞争和死锁情况。

实现简单的线程池

前置函数

Mutex 类是对 POSIX 线程互斥锁的封装，而 LockGuard 类则试图利用 RAII（Resource Acquisition Is Initialization）原则来自动管理锁的生命周期。

Mutex 类介绍

class Mutex
{
public:
Mutex(pthread_mutex_t* pMutex)
:_pMutex(pMutex)
{}
void Lock()
{pthread_mutex_lock(_pMutex);
}
void UnLock()
{pthread_mutex_unlock(_pMutex);
}~Mutex()
{}private:pthread_mutex_t* _pMutex;
};

构造函数：接收一个指向 pthread_mutex_t 的指针，并将其存储在私有成员 _pMutex 中。这个指针应该指向一个有效的、已经初始化的互斥锁。
Lock() 方法：调用 pthread_mutex_lock 函数来尝试锁定互斥锁。如果锁已经被其他线程持有，则当前线程会阻塞，直到锁变得可用。
UnLock() 方法：调用 pthread_mutex_unlock 函数来解锁互斥锁。解锁后，其他等待该锁的线程可以获得锁并执行其临界区的代码。
析构函数：目前为空，不执行任何操作。在实际应用中，如果互斥锁在 Mutex 对象销毁时仍然锁定，可能会导致问题（如死锁）。因此，一些实现可能会在析构函数中检查锁的状态，并尝试解锁（尽管这种做法有争议，因为它可能隐藏了编程错误）。

LockGuard 类介绍

class LockGuard
{
public:LockGuard(pthread_mutex_t* mutex):_mutex(mutex){_mutex.Lock();}~LockGuard(){_mutex.UnLock();}private:Mutex _mutex;
};

构造函数：接收一个指向 pthread_mutex_t 的指针，并使用该指针构造一个 Mutex 对象 _mutex。然后立即调用 _mutex.Lock() 来锁定互斥锁。这种方式确保了当 LockGuard 对象被创建时，相关的互斥锁会被立即锁定。
析构函数：在 LockGuard 对象被销毁时（例如，离开其作用域时）自动调用。析构函数调用 _mutex.UnLock() 来解锁互斥锁。这确保了无论何种情况下（包括异常），互斥锁都会被正确解锁，从而防止了死锁和其他多线程同步问题。
私有成员：_mutex 是一个 Mutex 类型的对象，它封装了对互斥锁的操作。由于 _mutex 是一个对象而非指针，我们不需要担心内存管理或空指针的问题。

Log类的介绍

#pragma once#include <iostream>
#include <fstream>
#include <string>
#include <cstdarg>
#include <ctime>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>enum
{Debug = 0,Info,Warning,Error,Fatal
};enum
{Screen = 10,OneFile,ClassFile
};std::string LevelToString(int level)
{switch (level){case Debug:return "Debug";case Info:return "Info";case Warning:return "Warning";case Error:return "Error";case Fatal:return "Fatal";default:return "Unknown";}
}const int defaultstyle = Screen;
const std::string default_filename = "log.";
const std::string logdir = "log";class Log
{
public:Log() : style(defaultstyle), filename(default_filename){mkdir(logdir.c_str(), 0775);}void Enable(int sty) //{style = sty;}std::string TimeStampExLocalTime(){time_t currtime = time(nullptr);struct tm *curr = localtime(&currtime);char time_buffer[128];snprintf(time_buffer, sizeof(time_buffer), "%d-%d-%d %d:%d:%d",curr->tm_year + 1900, curr->tm_mon + 1, curr->tm_mday,curr->tm_hour, curr->tm_min, curr->tm_sec);return time_buffer;}void WriteLogToOneFile(const std::string &logname, const std::string &message){umask(0);int fd = open(logname.c_str(), O_CREAT | O_WRONLY | O_APPEND, 0666);if (fd < 0)return;write(fd, message.c_str(), message.size());close(fd);}void WriteLogToClassFile(const std::string &levelstr, const std::string &message){std::string logname = logdir;logname += "/";logname += filename;logname += levelstr;WriteLogToOneFile(logname, message);}void WriteLog(const std::string &levelstr, const std::string &message){switch (style){case Screen:std::cout << message;break;case OneFile:WriteLogToClassFile("all", message);break;case ClassFile:WriteLogToClassFile(levelstr, message);break;default:break;}}void LogMessage(int level, const char *format, ...) // 类C的一个日志接口{char leftbuffer[1024];std::string levelstr = LevelToString(level);std::string currtime = TimeStampExLocalTime();std::string idstr = std::to_string(getpid());char rightbuffer[1024];va_list args; // char *, void *va_start(args, format);// args 指向了可变参数部分vsnprintf(rightbuffer, sizeof(rightbuffer), format, args);va_end(args); // args = nullptr;snprintf(leftbuffer, sizeof(leftbuffer), "[%s][%s][%s] ",levelstr.c_str(), currtime.c_str(), idstr.c_str());std::string loginfo = leftbuffer;loginfo += rightbuffer;WriteLog(levelstr, loginfo);}~Log() {}private:int style;std::string filename;
};Log lg;class Conf
{
public:Conf(){lg.Enable(ClassFile);}~Conf(){}
};Conf conf;

枚举定义

日志级别枚举：定义了五种日志级别，分别是Debug、Info、Warning、Error和Fatal。这些级别通常用于表示日志信息的重要性和紧急性。
日志输出方式枚举：定义了三种输出方式，Screen表示直接输出到屏幕，OneFile表示将所有日志写入同一个文件，ClassFile表示将不同级别的日志分别写入不同的文件。

Log类

这个类是日志系统的核心，它包含了以下方法和成员：

私有成员：
- style：表示当前的日志输出方式。
- filename：日志文件的默认名称。
构造函数：初始化日志系统和创建一个名为"log"的目录。
Enable方法：设置日志的输出方式。
TimeStampExLocalTime方法：返回当前的本地时间戳字符串。
WriteLogToOneFile方法：将日志信息写入指定的单个文件。
WriteLogToClassFile方法：根据日志级别将日志信息写入到对应的文件。
WriteLog方法：根据当前的日志输出方式，将日志信息输出到屏幕或文件。
LogMessage方法：这是一个可变参数的函数，用于格式化并记录日志信息。它接受一个日志级别和一个格式字符串，后面可以跟随任意数量的参数。这些参数会被格式化到日志信息中。
析构函数：目前为空，但可以在这里添加清理代码，如果需要的话。

全局对象

lg：是一个全局的Log对象，用于在整个程序中记录日志。
conf：是一个Conf对象，其构造函数中启用了ClassFile日志输出方式。这意味着，除非在程序的其他地方进行更改，否则日志将默认写入到分类的文件中。

Conf类

这个类目前非常简单，只包含一个构造函数和一个析构函数。构造函数中调用了lg.Enable(ClassFile)来设置日志的输出方式为分类文件输出。这个类可以用于未来扩展更多的配置选项。

myThread类的介绍

#pragma once#include <iostream>
#include <string>
#include <functional>
#include <pthread.h>
#include <string>template <class T>
using func_t = std::function<void(T&)>;template <class T>
class myThread
{
public:myThread(std::string threadName = "default", T data = T(), func_t<T> func = nullptr)//传入线程名、要操作的数据、要执行的函数: _tid(0),_threadName(threadName),_func(func),_data(data),_isRunning(false){}static void *ThreadRoutine(void *args)//内部的运行函数{myThread *pmt = static_cast<myThread *> (args);pmt->_func(pmt->_data);return nullptr;}bool Start()//创建进程{int ret = pthread_create(&_tid, NULL, ThreadRoutine, this);if (ret != 0){return false;}_isRunning = true;return true;}bool Join()//等待进程{if (!_isRunning){return true;}int ret = pthread_join(_tid, NULL);if (ret == 0){_isRunning = false;return true;}return false;}bool IsRunning()//返回进程的运行状态{return _isRunning;}const std::string& ThreadName()//返回线程名{return _threadName;}~myThread(){}private:pthread_t _tid;std::string _threadName;func_t<T> _func;T _data;bool _isRunning;
};

这个myThread类是一个C++模板类，用于封装POSIX线程（也称为pthreads）的创建、运行和等待过程。通过使用这个类，用户可以更方便地管理线程，而不需要直接处理底层的pthread API。下面是对这个类的详细介绍：

类成员变量

_tid: 存储线程ID的变量，类型为pthread_t。
_threadName: 一个字符串，用于存储线程的名称。
_func: 一个std::function对象，存储了要在线程中执行的函数。这个函数接受一个类型为T的引用作为参数。
_data: 一个类型为T的对象，它将被传递给_func函数。
_isRunning: 一个布尔变量，用于表示线程是否正在运行。

构造函数

构造函数myThread接受三个参数：线程名、要操作的数据以及要执行的函数。这些参数都有默认值，所以用户可以选择性地提供它们。

成员函数

ThreadRoutine: 这是一个静态成员函数，作为线程的入口点。它接受一个void*类型的参数（这是pthread API的要求），然后将其转换为myThread类的指针。接着，它调用存储在_func中的函数，并将_data作为参数传递。
Start: 这个函数用于创建并启动线程。它调用pthread_create函数，并传入ThreadRoutine作为线程的入口点。如果线程创建成功，它将_isRunning设置为true并返回true；否则返回false。
Join: 这个函数用于等待线程完成执行。如果线程已经在运行，它会调用pthread_join来等待线程结束。如果线程成功结束，它将_isRunning设置为false并返回true；否则返回false。如果线程没有运行，它直接返回true。
IsRunning: 这个函数返回一个布尔值，表示线程是否正在运行。
ThreadName: 这个函数返回线程的名称。
析构函数: 目前为空，但可以在这里添加必要的清理代码（例如，确保线程已经正确结束）。

使用场景

这个类适用于需要并发执行某个任务，同时又不希望直接处理复杂的线程API的场景。用户只需提供一个函数和一个数据对象，然后调用Start来创建和启动线程。当需要等待线程完成时，可以调用Join函数。

注意事项

由于这个类使用了C++11的特性（如std::function），因此需要确保编译器支持C++11或更高版本。
当使用多线程时，需要注意线程安全问题，特别是当多个线程访问共享数据时。
如果线程函数抛出异常，这个异常将不会被外部捕获，并可能导致程序崩溃。因此，需要确保线程函数不会抛出异常，或者在函数内部处理所有可能的异常。

Task类的介绍

#pragma once#include <iostream>
#include <string>enum ERROR
{Normal = 10,Div_Zeor,Mod_Zeor,UnKonwOperator
};std::string opers = "!@#$%^&*()_=+-*/";template <class T>
class Task
{
public:Task(){}Task(const T &data_x, const T &data_y, char oper): _data_y(data_y),_data_x(data_x),_oper(oper),_code(Normal),_result(0){}Task(const Task<T> &task): _data_x(task._data_x),_data_y(task._data_y),_oper(task._oper),_result(task._result),_code(task._code){}void Run(){switch (_oper){case '+':{_result = _data_x + _data_y;break;}case '-':{_result = _data_x - _data_y;break;}case '*':{_result = _data_x * _data_y;break;}case '/':{if (_data_y == 0){_code = Div_Zeor;break;}_result = _data_x / _data_y;break;}case '%':{if (_data_y == 0){_code = Mod_Zeor;break;}_result = _data_x % _data_y;break;}default:{_code = UnKonwOperator;break;}}}std::string Print(){std::string show;if (_code == Normal){show += "left[";show += std::to_string(_data_x);show += "]";show += _oper;show += "right[";show += std::to_string(_data_y);show += "]";show += "==";show += " ?";}else{show += "该任务非法";switch (_code){case Div_Zeor:show += "Div_Zeor";break;case Mod_Zeor:show += "Mod_Zeor";case UnKonwOperator:show += "UnKonwOperator";default:break;}}return show;}std::string PrintResult(){return std::to_string(_result);}private:T _data_x;T _data_y;T _result;char _oper;int _code;
};

这个Task类是一个模板类，设计用于执行两个同类型数据之间的基本算术运算。模板类型T允许这个类处理不同的数据类型，只要这些类型支持算术运算符。这个类的主要功能包括：

成员变量

_data_x 和 _data_y：存储要执行运算的两个操作数。
_oper：存储要执行的运算符（如+, -, *, /, %）。
_result：存储运算的结果。
_code：存储错误码，用于表示运算过程中是否出现错误（如除以零错误或未知运算符）。

成员函数

构造函数：有三个构造函数，一个默认构造函数，一个接受两个数据和一个运算符作为参数的构造函数，以及一个拷贝构造函数。
**Run**：执行实际的运算。根据_oper的值执行相应的算术运算，并将结果存储在_result中。如果运算过程中出现错误（如除以零），则更新_code以反映错误类型。
**Print**：返回一个字符串，表示运算的详细信息和可能发生的错误。如果没有错误，它将返回一个表示运算的字符串（例如，"left[5]+right[3]== ?"）。如果发生错误，它将包含错误类型（例如，"该任务非法Div_Zeor"）。
**PrintResult**：返回表示运算结果的字符串。

错误处理

ERROR枚举用于定义可能的错误类型，如Div_Zeor（除数为零）、Mod_Zeor（取模运算的除数为零）和UnKonwOperator（未知运算符）。这些错误码在运算过程中被设置，并可以通过Print函数来查看。

使用场景

这个类可以用于创建一个简单的算术表达式求值器，能够处理基本的算术运算。由于它是一个模板类，因此可以很容易地处理不同类型的数据，如整数、浮点数等。

注意事项

这个类没有进行复杂的错误处理，例如处理不支持的数据类型或检查运算符的有效性（除了基本的算术运算符外）。
在多线程环境中使用这个类时需要注意线程安全，因为这个类不是线程安全的。
如果需要处理更复杂的算术表达式（例如，包含括号、多个运算符和函数的表达式），则需要一个更复杂的解析器和求值器。

线程池的实现

#pragma once#include <iostream>
#include <queue>
#include <vector>
#include <string>
#include "Task.hpp"
#include "LockGuard.hpp"
#include "myThread.hpp"
#include "Log.hpp"
#include <pthread.h>const int Deafult_Num = 10;
struct ThreadData
{std::string _threaName;ThreadData(const std::string &threadname): _threaName(threadname){}~ThreadData(){}
};template <class T>
class ThreadPool
{
private:ThreadPool(const int threadNum = Deafult_Num): _threadNum(threadNum){pthread_cond_init(&_cond, nullptr);pthread_mutex_init(&_mutex, nullptr);for (int i = 0; i < _threadNum; i++){std::string threadName = "thread-";threadName += std::to_string(i + 1);ThreadData tmp(threadName);myThread<ThreadData> tmp_thread(threadName, tmp, std::bind(&ThreadPool::ThreadRun, this, std::placeholders::_1));_threads.emplace_back(tmp_thread);lg.LogMessage(Info, "%s is created...\n", threadName.c_str());}}ThreadPool(const ThreadPool<T> &tp) = delete;const ThreadPool<T> &operator=(const ThreadPool<T>) = delete;public:// 有线程安全问题的static ThreadPool<T> *GetInstance(){if (instance == nullptr){LockGuard lockguard(&sig_lock);if (instance == nullptr){lg.LogMessage(Info, "创建单例成功...\n");instance = new ThreadPool<T>();}}return instance;}bool Start(){// 启动for (auto &thread : _threads){thread.Start();lg.LogMessage(Info, "%s is running ...\n", thread.ThreadName().c_str());}return true;}void ThreadRun(ThreadData &td){while (true){// 取任务T t;{LockGuard lockguard(&_mutex);while (_queue.empty()){ThreadWait(td);lg.LogMessage(Debug, "thread %s is wakeup\n", td._threaName.c_str());}t = _queue.front();_queue.pop();}// 处理任务t.Run();lg.LogMessage(Debug, "%s handler task %s done, result is : %s\n",td._threaName.c_str(), t.Print().c_str(), t.PrintResult().c_str());}}void ThreadWait(const ThreadData &td){lg.LogMessage(Debug, "no task, %s is sleeping...\n", td._threaName.c_str());pthread_cond_wait(&_cond, &_mutex);}void ThreadWakeup(){pthread_cond_signal(&_cond);}void Push(T &in){lg.LogMessage(Debug, "other thread push a task, task is : %s\n", in.Print().c_str());LockGuard lockguard(&_mutex);_queue.push(in);ThreadWakeup();}void Wait(){for (auto &thread : _threads){thread.Join();}}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}
private:std::queue<T> _queue;std::vector<myThread<ThreadData>> _threads;int _threadNum;pthread_cond_t _cond;pthread_mutex_t _mutex;static ThreadPool<T> *instance;static pthread_mutex_t sig_lock;
};template <class T>
ThreadPool<T> *ThreadPool<T>::instance = nullptr;
template <class T>
pthread_mutex_t ThreadPool<T>::sig_lock = PTHREAD_MUTEX_INITIALIZER;

ThreadPool 类

ThreadPool是一个线程池模板类，它允许你并行地处理一系列的任务。线程池中的线程数量可以在创建时指定，或者使用默认值（在这个例子中是Deafult_Num，尽管这个单词应该是Default_Num的拼写错误）。这个类使用了POSIX线程库（pthread）来实现线程同步。

成员变量

private:std::queue<T> _queue;std::vector<myThread<ThreadData>> _threads;int _threadNum;pthread_cond_t _cond;pthread_mutex_t _mutex;static ThreadPool<T> *instance;static pthread_mutex_t sig_lock;

_queue：一个存储Task类型对象的队列。
_threads：一个包含myThread<ThreadData>对象的向量，表示线程池中的线程。
_threadNum：线程池中的线程数量。
_cond：一个条件变量，用于线程的等待和唤醒。
_mutex：一个互斥锁，用于保护共享资源（如任务队列）。
instance：单例模式的实例指针。
sig_lock：保护单例实例化的互斥锁。

成员函数

构造函数 (ThreadPool)

class ThreadPool
{
private:ThreadPool(const int threadNum = Deafult_Num): _threadNum(threadNum){pthread_cond_init(&_cond, nullptr);pthread_mutex_init(&_mutex, nullptr);for (int i = 0; i < _threadNum; i++){std::string threadName = "thread-";threadName += std::to_string(i + 1);ThreadData tmp(threadName);myThread<ThreadData> tmp_thread(threadName, tmp, std::bind(&ThreadPool::ThreadRun, this, std::placeholders::_1));_threads.emplace_back(tmp_thread);lg.LogMessage(Info, "%s is created...\n", threadName.c_str());}}

* 初始化线程池，设置线程数量，初始化条件变量和互斥锁。
* 根据线程数量创建相应数量的线程，并存储在`_threads`向量中。
* 线程名通过`ThreadData`结构传递，并记录日志。

GetInstance

  static ThreadPool<T> *GetInstance(){if (instance == nullptr){LockGuard lockguard(&sig_lock);if (instance == nullptr){lg.LogMessage(Info, "创建单例成功...\n");instance = new ThreadPool<T>();}}return instance;}

* 实现单例模式，确保整个程序中只有一个`ThreadPool`实例。
* 使用双重检查锁定（Double-Checked Locking）来确保线程安全。

Start

  bool Start(){// 启动for (auto &thread : _threads){thread.Start();lg.LogMessage(Info, "%s is running ...\n", thread.ThreadName().c_str());}return true;}

* 启动线程池中的所有线程。
* 记录每个线程启动的日志。

ThreadRun

  void ThreadRun(ThreadData &td){while (true){// 取任务T t;{LockGuard lockguard(&_mutex);while (_queue.empty()){ThreadWait(td);lg.LogMessage(Debug, "thread %s is wakeup\n", td._threaName.c_str());}t = _queue.front();_queue.pop();}// 处理任务t.Run();lg.LogMessage(Debug, "%s handler task %s done, result is : %s\n",td._threaName.c_str(), t.Print().c_str(), t.PrintResult().c_str());}}

* 线程的工作函数，每个线程都会执行这个函数。
* 在一个无限循环中，线程从任务队列中取出一个任务，执行它，并记录日志。
* 如果任务队列为空，线程会等待，直到有新的任务被推入队列。

ThreadWait

  void ThreadWait(const ThreadData &td){lg.LogMessage(Debug, "no task, %s is sleeping...\n", td._threaName.c_str());pthread_cond_wait(&_cond, &_mutex);}

* 当任务队列为空时，线程会调用这个函数来等待。
* 使用条件变量来挂起线程，直到有新的任务被推入队列。

ThreadWakeup

  void ThreadWakeup(){pthread_cond_signal(&_cond);}

* 当有新任务被推入队列时，调用此函数来唤醒一个等待的线程。
* 使用条件变量的信号功能来实现。

Push

  void Push(T &in){lg.LogMessage(Debug, "other thread push a task, task is : %s\n", in.Print().c_str());LockGuard lockguard(&_mutex);_queue.push(in);ThreadWakeup();}

* 允许其他线程将一个任务推入线程池的任务队列。
* 使用互斥锁来保护任务队列的线程安全。
* 唤醒一个等待的线程来处理新任务。

Wait

  void Wait(){for (auto &thread : _threads){thread.Join();}}

* 等待线程池中的所有线程完成它们当前的任务并退出。
* 通常用于程序的结束阶段，确保所有任务都被处理完。

**析构函数

  ~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}

* 析构函数应该负责清理资源，如销毁线程、销毁互斥锁和条件变量等。
* 还需要注意线程安全地停止和销毁线程。