当前位置：首页 > news >正文

线程池及线程池单例模式

news 2025/8/29 9:10:53

线程池设计

线程池：

线程池的应用场景：

线程池的种类：

线程池的实现

线程池的私有成员:

对类中各个成员函数的理解

线程池构造函数ThreadPool()

HandlerTask()

前言

正文

Stop()

Enqueue()

单例模式线程池

什么是单例模式？

饿汉方式实现单例模式

懒汉方式实现单例模式

实现单例的关键步骤：

1. 构造函数私有化

2. 禁止拷贝和赋值

3. 静态实例指针

4. 提供全局访问方法

使用示例：

单例线程池完整代码

线程池设计

线程池：

一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

线程池的应用场景：

需要大量的线程来完成任务，且完成任务的时间比较短。比如WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误。

线程池的种类：

a. 创建固定数量线程池，循环从任务队列中获取任务对象，获取到任务对象后，执行任务对象中的任务接口
b. 浮动线程池，其他同上

此处，我们选择固定线程个数的线程池。

线程池的实现

#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <queue>
#include "Log.hpp"
#include "Thread.hpp"
#include "Cond.hpp"
#include "Mutex.hpp"// .hpp header onlynamespace ThreadPoolModule
{using namespace ThreadModlue;using namespace LogMudoule;using namespace CondModule;using namespace MutexModule;static const int gnum = 5;template <typename T>class ThreadPool{private:void WakeUpOne(){_cond.Signal();LOG(LogLevel::INFO) << "唤醒一个休眠线程";}void WakeUpAllThread(){LockGuard lockguard(_mutex);if (_sleepernum)_cond.Broadcast();LOG(LogLevel::INFO) << "唤醒所有的休眠线程";}void HandlerTask(){char name[128];pthread_getname_np(pthread_self(), name, sizeof(name));while (true){T t;{LockGuard lockguard(_mutex);// 1. a.队列为空 b. 线程池没有退出while (_taskq.empty() && _isrunning){_sleepernum++;_cond.Wait(_mutex);_sleepernum--;}// 2. 内部的线程被唤醒if (!_isrunning && _taskq.empty()){LOG(LogLevel::INFO) << name << " 退出了, 线程池退出&&任务队列为空";break;}t=_taskq.front();_taskq.pop();}// t();}}public:ThreadPool(int num = gnum) : _num(num), _isrunning(false), _sleepernum(0){for (int i = 0; i < num; i++){_threads.emplace_back([this](){HandlerTask();});}}void Start(){if (_isrunning){return;}_isrunning = true;for (auto &thread : _threads){thread.Start();LOG(LogLevel::INFO) << "start new thread success: " << thread.Name();}}void Stop(){if (!_isrunning) // 线程结束就返回return;_isrunning = false;// 唤醒所有的线层WakeUpAllThread();}void Join(){for (auto &thread : _threads){thread.Join();}}bool Enqueue(const T &in){if (_isrunning){LockGuard lockguard(_mutex);_taskq.push(in);if (_threads.size() == _sleepernum)WakeUpOne();return true;}return false;}~ThreadPool(){}private:std::vector<Thread> _threads;int _num;std::queue<T> _taskq;Cond _cond;Mutex _mutex;bool _isrunning;int _sleepernum;};}

效果展示：

线程池的私有成员:

线程线程个数任务队列条件变量互斥锁判断线程是否运行线程等待(休眠)的个数

这些成员不难理解。

这里为什么要有互斥锁和条件变量呢？

线程池中的任务队列是会被多个执行流同时访问的临界资源，因此我们需要引入互斥锁对任务队列进行保护。

线程池当中的线程要从任务队列里拿任务，前提条件是任务队列中必须要有任务，因此线程池当中的线程在拿任务之前，需要先判断任务队列当中是否有任务，若此时任务队列为空，那么该线程应该进行等待，直到任务队列中有任务时再将其唤醒，因此我们需要引入条件变量。

当外部线程向任务队列中Push一个任务后，此时可能有线程正处于等待状态，因此在新增任务后需要唤醒在条件变量下等待的线程。

对类中各个成员函数的理解

线程池构造函数ThreadPool()

ThreadPool(int num = gnum) : _num(num), _isrunning(false), _sleepernum(0)
{for (int i = 0; i < num; i++){_threads.emplace_back([this](){HandlerTask();});}
}

_num(num)：记录线程数量
_isrunning(false)：标记线程池尚未开始运行
_sleepernum(0)：初始休眠线程数为0

循环创建num个工作线程
使用emplace_back向_threads对象中添加线程对象
每个线程都执行相同的lambda函数，该函数捕获this指针并调用HandlerTask()方法

HandlerTask()

void HandlerTask()
{char name[128];pthread_getname_np(pthread_self(), name, sizeof(name));while (true){T t;{LockGuard lockguard(_mutex);// 1. a.队列为空 b. 线程池没有退出while (_taskq.empty() && _isrunning){_sleepernum++;_cond.Wait(_mutex);_sleepernum--;}// 2. 内部的线程被唤醒if (!_isrunning && _taskq.empty()){LOG(LogLevel::INFO) << name << " 退出了, 线程池退出&&任务队列为空";break;}t = _taskq.front();_taskq.pop();}// t();}
}

前言

HandlerTask()是怎么通过回调函数Routinue执行的？

由于我们封装了Thread.hpp，在线程类里HandlerTask()是通过回调函数Routine执行的：

// 在线程池构造函数中创建线程
_threads.emplace_back([this](){ HandlerTask(); });

Thread.hpp中Thread类的Start()方法

bool Start()
{int n = pthread_create(&_tid, nullptr, Routine, this);// ...
}

创建POSIX线程，指定Routine为线程入口函数将this指针（Thread对象）作为参数传递给Routine

Routine函数（静态成员函数）

static void *Routine(void *args)
{Thread *self = static_cast<Thread *>(args);self->EnableRunning();if (self->_isdetach)self->Detach();pthread_setname_np(self->_tid, self->_name.c_str());self->_func(); // 这里调用HandlerTask()!!!return nullptr;
}

_func是在Thread构造函数中设置的lambda函数，当self->_func()执行时，实际上就是调用：this->HandlerTask();

完整调用链：

pthread_create() → Routine() → self->_func() → HandlerTask()

正文

任务的处理是属于临界资源，任务队列需要上锁的主要原因是为了保证线程安全，防止多个线程同时访问和修改队列时出现数据竞争和不一致性问题。
当某线程被唤醒时，其可能是被异常或是伪唤醒，或者是一些广播类的唤醒线程操作而导致所有线程被唤醒，使得在被唤醒的若干线程中，只有个别线程能拿到任务。此时应该让被唤醒的线程再次判断是否满足被唤醒条件，所以在判断任务队列是否为空时，应该使用while进行判断，而不是if
在判断队列是否为空时，我们默认期望的是线程正在运行，所以当任务队列为空，线程来了就只能休眠。走到后面(if后)证明任务队列不为空且线程被唤醒，我们就取出任务并pop。当我们break时一定是任务队列为空，且线程已经结束了<-if的判断条件

Stop()

void Stop()
{if (!_isrunning) // 线程结束就返回return;_isrunning = false;// 唤醒所有的线层WakeUpAllThread();
}

在Stop函数中我们设置判断线程运行的标志位为false即结束状态

然后我们要唤醒所有线程，如果不唤醒所有线程，线程会在HandlerTask函数中一直休眠。

然后通过判断如果任务队列为空了，就会退出循环线程break结束。

Enqueue()

bool Enqueue(const T &in)
{if (_isrunning){LockGuard lockguard(_mutex); // 上锁保护共享资源_taskq.push(in);// 关键判断：如果所有线程都在休眠，就唤醒一个if (_threads.size() == _sleepernum)WakeUpOne();return true;}return false;
}

任务入队函数，用于向任务队列添加新任务

其他的不在详细述说，接下来我们把线程池改造成单例模式

单例模式线程池

什么是单例模式？

某些类,只应该具有⼀个对象(实例)，就称之为单例.。例如⼀个男人只能有⼀个媳妇.，在很多服务器开发场景中,经常需要让服务器加载很多的数据(上百G)到内存中，此时往往要用⼀个单例的类来管理这些数据.

饿汉方式实现单例模式

template <typename T>
class Singleton
{static T data;public:static T *GetInstance(){return &data;}
};

只要通过Singleton这个包装类来使用T对象,则⼀个进程中只有⼀个T对象的实例

懒汉方式实现单例模式

template <typename T>
class Singleton
{static T *inst;public:static T *GetInstance(){if (inst == NULL){inst = new T();}return inst;}
};

存在⼀个严重的问题,线程不安全. 第⼀次调用GetInstance的时候,如果两个线程同时调用,可能会创建出两份T对象的实例. 但是后续再次调用,就没有问题了

// 懒汉模式, 线程安全
template <typename T>
class Singleton
{volatile static T *inst; // 需要设置 volatile 关键字, 否则可能被编译器优化static std::mutex lock;public:static T *GetInstance(){if (inst == NULL){lock.lock();if (inst == NULL){inst = new T();}lock.unlock();}return inst;}
};

注意事项:

加锁解锁的位置
双重if判定,避免不必要的锁竞争
volatile关键字防止过度优化