一步一步写线程之八线程池的完善之一线程管理

2024-03-24 02:44
文章标签 线程 管理 完善 一步 之八

本文主要是介绍一步一步写线程之八线程池的完善之一线程管理，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、介绍

在前面实现了一个很粗陋的线程池，那么它的运行没有动态控制，相关线程池的动态伸缩也没有控制，这一切，在实际的应用中，是应该存在的。本篇文章就将相关的实现方法和源码分析一下。

二、线程池内线程的管理

线程池中的线程管理，一般有以下几种：
1、动态控制线程的启停
这意味着线程在启动时，不应用直接启动，而是根据实际情况来启动。
2、线程池的容量管理
线程的创建数量其实和实际场景中的硬件资源和OS系统强相关，所以线程池中的线程创建多少，需要开发者一开始设置一个处置的上限。达到既可以利用多线程的优势，又不能脱离实际资源以及浪费大量的时间在线程的上下文切换上。
3、获取线程ID
这个相对来说比较简单，STL中提供了相关的接口函数。
4、动态扩容和收缩
这个相对来说实现容易管理难。比如扩容和收缩的机制，如何才能更好的适应实际情况。

三、例程

先看一下代码，主要看一下与上一篇的不同之处：

//common.h#ifndef __COMMON_H__
#define __COMMON_H__#include <functional>using CallBackMsg = std::function<void(int *, int)>;
using Task = std::function<void(int)>;class NoCopy {
protected:NoCopy() = default;~NoCopy() = default;public:NoCopy(const NoCopy &) = delete;NoCopy &operator=(const NoCopy &) = delete;NoCopy(NoCopy &&) = delete;NoCopy &operator=(NoCopy &&) = delete;
};#endif // __COMMON_H__
//ThreadCondition.h
#include <condition_variable>
#include <mutex>#include "common.h"class ThreadCondition : private NoCopy {
public:ThreadCondition() {}~ThreadCondition() {}public:inline bool Wait(int timeOut) {signaled_ = false;std::unique_lock<std::mutex> lock(this->lockMutex_);if (this->cvLock_.wait_for(lock, std::chrono::milliseconds(timeOut)) == std::cv_status::timeout) {return false;}return true;}inline void Wait() {signaled_ = false;std::unique_lock<std::mutex> lock(this->lockMutex_);while (!signaled_) {this->cvLock_.wait(lock);}}inline void Signal() {std::unique_lock<std::mutex> lock(this->lockMutex_);signaled_ = true;// pthread_cond_broadcast(&_cond);this->cvLock_.notify_one();}void SetSignal(bool quit = false) noexcept {//设置退出循环标志if (quit) {this->quit_ = true;}//唤醒线程this->Signal();}private:bool signaled_ = false;std::mutex lockMutex_;std::condition_variable cvLock_;bool quit_ = false;
};//ThreadPool.h
#ifndef __THREADPOOL_H__
#define __THREADPOOL_H__#include "TaskQueue.h"
#include "ThreadCondition.h"
#include "common.h"
#include <atomic>
#include <memory>
#include <thread>
#include <vector>class ThreadWorker;class ThreadPool {
public:explicit ThreadPool();~ThreadPool();public:void InitThreadPool(int threadCount, bool initThreadStatus, CallBackMsg cb);void AddTask(Task t);std::tuple<bool, Task> GetTask();void Extend(int num = 3);void Contact(int num = 3);public:void Wait();void SetSignal();void SetSignal(std::thread::id threadid);void SetMaxThreadCount(int maxCount);std::vector<std::thread::id> GetThreadID();public:static std::shared_ptr<ThreadPool> Get();private:void Destory();bool getThreadRunStatus(std::thread::id id);private:std::vector<std::shared_ptr<ThreadWorker>> pVecThreadWorker_;CallBackMsg funcCallBack_ = nullptr;std::atomic<int> curId_ = 0;TaskQueue<Task> taskQueue_;std::shared_ptr<ThreadCondition> pThreadCon_ = nullptr;bool initThreadStatus_ = false;int threadCount_ = 0;int maxThreadCount_ = 30;std::vector<std::thread::id> idVec_;
};
#endif // __THREADPOOL_H__//ThreadPool.cpp
#include "ThreadPool.h"
#include "ThreadWorker.h"
#include <iostream>ThreadPool::ThreadPool() {}
ThreadPool::~ThreadPool() { this->Destory(); }
void ThreadPool::InitThreadPool(int threadCount, bool initThreadStatus, CallBackMsg cb) {this->funcCallBack_ = cb;if (threadCount > this->maxThreadCount_) {threadCount = this->maxThreadCount_;}this->threadCount_ = threadCount;this->initThreadStatus_ = initThreadStatus;for (int num = 0; num < threadCount; num++) {auto workerThread = std::make_shared<ThreadWorker>();workerThread->InitThread(initThreadStatus, cb);this->pVecThreadWorker_.emplace_back(workerThread);}this->pThreadCon_ = std::make_shared<ThreadCondition>();
}
void ThreadPool::Extend(int num) {for (int num = 0; num < num; num++) {auto workerThread = std::make_shared<ThreadWorker>();workerThread->InitThread(initThreadStatus_, this->funcCallBack_);this->pVecThreadWorker_.emplace_back(workerThread);}
}
void ThreadPool::Contact(int num) {int index = this->threadCount_ - 1;for (int id = 0; id < num; id++) {//auto th = this->pVecThreadWorker_[index--];th->Quit();this->pVecThreadWorker_.pop_back();}
}
void ThreadPool::AddTask(Task t) {this->taskQueue_.Push(t);this->SetSignal();std::cerr << "add task and signal" << std::endl;
}
void ThreadPool::Destory() {for (auto &au : this->pVecThreadWorker_) {au->Join();}
}
std::tuple<bool, Task> ThreadPool::GetTask() { return this->taskQueue_.PopFront(); }
std::shared_ptr<ThreadPool> ThreadPool::Get() {static auto threadPool = std::make_shared<ThreadPool>();return threadPool;
}void ThreadPool::Wait() { this->pThreadCon_->Wait(); }
void ThreadPool::SetSignal() { this->pThreadCon_->Signal(); }
// Wake up specified thread
void ThreadPool::SetSignal(std::thread::id threadid) {for (auto &pWorker : this->pVecThreadWorker_) {if (pWorker->GetCurThreadID() == threadid) {pWorker->Start();}}
}
void ThreadPool::SetMaxThreadCount(int maxCount) { this->maxThreadCount_ = maxCount; }
bool ThreadPool::getThreadRunStatus(std::thread::id id) {for (auto &worker : this->pVecThreadWorker_) {if (id == worker->GetCurThreadID()) {return worker->getCurRunStatus();}}return false;
}
std::vector<std::thread::id> ThreadPool::GetThreadID() {for (auto &worker : this->pVecThreadWorker_) {this->idVec_.emplace_back(worker->GetCurThreadID().value());}return this->idVec_;
}//ThreadWorker.h
#ifndef __THREADWORKER_H__
#define __THREADWORKER_H__#include "ThreadCondition.h"
#include "common.h"
#include <atomic>
#include <memory>
#include <optional>
#include <thread>class ThreadWorker {
public:ThreadWorker();~ThreadWorker() = default;public:void InitThread(bool initStatus, CallBackMsg cb);void Start();void SetSignal();void Run();void Join();void Quit();bool getCurRunStatus();std::optional<std::thread::id> GetCurThreadID() { return this->curThreadId_; }private:std::shared_ptr<std::thread> pWorkerThread_ = nullptr;std::atomic<bool> status_ = false;bool runStatus_ = false;std::thread::id curThreadId_;// std::shared_ptr<ThreadCondition> pThreadCon_ = nullptr;CallBackMsg cbm_;
};#endif // __THREADWORKER_H__
//ThreadWorker.cpp
#include "ThreadWorker.h"
#include "ThreadPool.h"
#include <iostream>ThreadWorker::ThreadWorker() {}void ThreadWorker::InitThread(bool initStatus, CallBackMsg cb) {this->cbm_ = cb;this->pWorkerThread_ = std::make_shared<std::thread>(&ThreadWorker::Run, this);if (nullptr != this->pWorkerThread_) {this->curThreadId_ = this->pWorkerThread_->get_id();}// this->pThreadCon_ = std::make_shared<ThreadCondition>();
}
void ThreadWorker::Start() {}
// void ThreadWorker::SetSignal() { this->pThreadCon_->Signal(); }
void ThreadWorker::Run() {int data[10] = {0};auto tPool = ThreadPool::Get();while (!status_) {this->runStatus_ = false;// this->pThreadCon_->Wait(); // Handling false wake-upstd::cerr << "cur thread id is:" << this->curThreadId_ << std::endl;ThreadPool::Get()->Wait();this->runStatus_ = true;std::cerr << "cur run thread id is:" << this->curThreadId_ << std::endl;auto [bRet, t] = tPool->GetTask();if (bRet) {t(10);data[0] = 10;data[1] = 11;data[2] = 12;cbm_(data, 3);} else {std::this_thread::sleep_for(std::chrono::milliseconds(100));if (this->cbm_ != nullptr) {cbm_(data, 0);}}std::this_thread::yield();}
}void ThreadWorker::Join() {if (this->pWorkerThread_ != nullptr && this->pWorkerThread_->joinable()) {this->pWorkerThread_->join();}
}void ThreadWorker::Quit() {this->status_ = true;// trigger conditional//   this->pThreadCon_.Signal();
}
bool ThreadWorker::getCurRunStatus() { return this->runStatus_; }

上面的代码有一些需要商榷的地方，比如现在是把条件变量的触发放到线程池的类中，所有线程都使用这一个，这样会不会引起类似惊群的现象？在线程池收缩时，线程退出时，是设置好线程状态好，再noticeAll，还是一个个的触发退出好。所以那块目前把代码注释掉了。如果使用触发所有，还需要在条件变量的类中增加noticeAll的相关API接口。还有如果收缩时退出线程正在运行怎么处理？这些都需要不断的完善。所以很多东西大家看别人的代码时，其实是理解不全面很多东西的，可自己亲自撸一把代码，思路就清晰很多。

四、分析

在上面的实现中，其实仍然还可以优化，比如将ThreadConditon类封装到每个ThreadWorker中，这样就可以精确控制每个线程的启停（Google的一个开源的框架中就是使用了类似的机制），这也为实现LF线程池的模型提供基础准备；另外，线程的退出机制需要完善，如何退出指定线程?是不是需要增加当前线程动态数量（多少运行，多少等待以及整体多少数量等待）的获取API，当前线程运行状态与收缩时的动态匹配等等。
其实上面的实现的也没有多么复杂，即使如此，线程池中的线程并行化控制仍然有很多没有解决，比如线程与CPU核心的绑定，线程的分组，线程的负载平衡（即防止线程的局部化应用，有的线程反复工作，有的线程发生饥饿现象）、条件变量的模板化封装等等。这个在后续逐渐完善，大家不要着急。

五、总结

在前面的文章发出去后，有兄弟留言说这个太简陋，对实际应用没啥作用。其实这个系列的文章重点不是一开始就推出一个完善的线程池供大家使用，更多的给初学写线程池的人一个设计编写的实操过程。正如前面反复说的，一个好的线程池一定是适应一种或少数几种场景的而不是普适于所有场景。所以重要是让大家有一个理解网上开源的线程池是如何写出来的过程。只要明白了设计和实现线程池的过程，那么这个线程池如何写好，就看个人的水平了。否则的话，直接从网上下载一个线程池用就可以了。可这样，又怎么能够针对实际应用的不同的场景进行完善呢？肯定还是要吃透别人的代码。而这个系列文章就是让大家在看别人的代码时，比较轻松的搞定相关的框架流程。
本篇的线程管理的完善，其实就是可以当成一种逐步对线程池完善的开始，如何管理线程池里的线程，甚至精确管理每一个线程，都可以根据情况来定。
罗马不是一天建成的，慢慢来，每天进步一点点！

这篇关于一步一步写线程之八线程池的完善之一线程管理的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/840250。

相关文章

SpringBoot 多环境开发实战(从配置、管理与控制)

SpringBoot 多环境开发实战(从配置、管理与控制)

《SpringBoot多环境开发实战(从配置、管理与控制)》本文详解SpringBoot多环境配置,涵盖单文件YAML、多文件模式、MavenProfile分组及激活策略,通过优先级控制灵活切换环境... 目录一、多环境开发基础（单文件 YAML 版）（一）配置原理与优势（二）实操示例二、多环境开发多文件版
阅读更多...
Java中如何正确的停掉线程

Java中如何正确的停掉线程

《Java中如何正确的停掉线程》Java通过interrupt()通知线程停止而非强制,确保线程自主处理中断,避免数据损坏,线程池的shutdown()等待任务完成,shutdownNow()强制中断... 目录为什么不强制停止为什么 Java 不提供强制停止线程的能力呢？如何用interrupt停止线程s
阅读更多...
Redis实现高效内存管理的示例代码

Redis实现高效内存管理的示例代码

《Redis实现高效内存管理的示例代码》Redis内存管理是其核心功能之一,为了高效地利用内存,Redis采用了多种技术和策略,如优化的数据结构、内存分配策略、内存回收、数据压缩等,下面就来详细的介绍... 目录1. 内存分配策略jemalloc 的使用2. 数据压缩和编码ziplist示例代码3. 优化的
阅读更多...
SpringBoot集成XXL-JOB实现任务管理全流程

SpringBoot集成XXL-JOB实现任务管理全流程

《SpringBoot集成XXL-JOB实现任务管理全流程》XXL-JOB是一款轻量级分布式任务调度平台,功能丰富、界面简洁、易于扩展,本文介绍如何通过SpringBoot项目,使用RestTempl... 目录一、前言二、项目结构简述三、Maven 依赖四、Controller 代码详解五、Service
阅读更多...
深入解析C++ 中std::map内存管理

深入解析C++ 中std::map内存管理

《深入解析C++中std::map内存管理》文章详解C++std::map内存管理,指出clear()仅删除元素可能不释放底层内存,建议用swap()与空map交换以彻底释放,针对指针类型需手动de... 目录1️、基本清空std::map2️、使用 swap 彻底释放内存3️、map 中存储指针类型的对象
阅读更多...
python 线程池顺序执行的方法实现

python 线程池顺序执行的方法实现

《python线程池顺序执行的方法实现》在Python中,线程池默认是并发执行任务的,但若需要实现任务的顺序执行,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋... 目录方案一：强制单线程（伪顺序执行）方案二：按提交顺序获取结果方案三：任务间依赖控制方案四：队列顺序消
阅读更多...
Linux系统管理与进程任务管理方式

Linux系统管理与进程任务管理方式

《Linux系统管理与进程任务管理方式》本文系统讲解Linux管理核心技能,涵盖引导流程、服务控制（Systemd与GRUB2）、进程管理（前台/后台运行、工具使用）、计划任务（at/cron）及常用... 目录引言一、linux系统引导过程与服务控制1.1 系统引导的五个关键阶段1.2 GRUB2的进化优
阅读更多...
Spring Security 前后端分离场景下的会话并发管理

Spring Security 前后端分离场景下的会话并发管理

《SpringSecurity前后端分离场景下的会话并发管理》本文介绍了在前后端分离架构下实现SpringSecurity会话并发管理的问题,传统Web开发中只需简单配置sessionManage... 目录背景分析传统 web 开发中的 sessionManagement 入口ConcurrentSess
阅读更多...
SpringBoot实现虚拟线程的方案

SpringBoot实现虚拟线程的方案

《SpringBoot实现虚拟线程的方案》Java19引入虚拟线程,本文就来介绍一下SpringBoot实现虚拟线程的方案,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,... 目录什么是虚拟线程虚拟线程和普通线程的区别SpringBoot使用虚拟线程配置@Async性能对比H
阅读更多...
Linux之UDP和TCP报头管理方式

Linux之UDP和TCP报头管理方式

《Linux之UDP和TCP报头管理方式》文章系统讲解了传输层协议UDP与TCP的核心区别：UDP无连接、不可靠,适合实时传输（如视频）,通过端口号标识应用；TCP有连接、可靠,通过确认应答、序号、窗... 目录一、关于端口号1.1 端口号的理解1.2 端口号范围的划分1.3 认识知名端口号1.4 一个进程
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2