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

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中使用 ThreadLocal 进行多线程上下文管理及注意事项小结

SpringBoot中使用 ThreadLocal 进行多线程上下文管理及注意事项小结

《SpringBoot中使用ThreadLocal进行多线程上下文管理及注意事项小结》本文详细介绍了ThreadLocal的原理、使用场景和示例代码,并在SpringBoot中使用ThreadLo... 目录前言技术积累1.什么是 ThreadLocal2. ThreadLocal 的原理2.1 线程隔离2
阅读更多...
Java多线程父线程向子线程传值问题及解决

Java多线程父线程向子线程传值问题及解决

《Java多线程父线程向子线程传值问题及解决》文章总结了5种解决父子之间数据传递困扰的解决方案,包括ThreadLocal+TaskDecorator、UserUtils、CustomTaskDeco... 目录1 背景2 ThreadLocal+TaskDecorator3 RequestContextH
阅读更多...
java父子线程之间实现共享传递数据

java父子线程之间实现共享传递数据

《java父子线程之间实现共享传递数据》本文介绍了Java中父子线程间共享传递数据的几种方法,包括ThreadLocal变量、并发集合和内存队列或消息队列,并提醒注意并发安全问题... 目录通过 ThreadLocal 变量共享数据通过并发集合共享数据通过内存队列或消息队列共享数据注意并发安全问题总结在 J
阅读更多...
异步线程traceId如何实现传递

异步线程traceId如何实现传递

《异步线程traceId如何实现传递》文章介绍了如何在异步请求中传递traceId,通过重写ThreadPoolTaskExecutor的方法和实现TaskDecorator接口来增强线程池,确保异步... 目录前言重写ThreadPoolTaskExecutor中方法线程池增强总结前言在日常问题排查中，
阅读更多...
Linux内存泄露的原因排查和解决方案（内存管理方法）

Linux内存泄露的原因排查和解决方案（内存管理方法）

《Linux内存泄露的原因排查和解决方案（内存管理方法）》文章主要介绍了运维团队在Linux处理LB服务内存暴涨、内存报警问题的过程,从发现问题、排查原因到制定解决方案,并从中学习了Linux内存管理... 目录一、问题二、排查过程三、解决方案四、内存管理方法1）linux内存寻址2）Linux分页机制3）
阅读更多...
高效管理你的Linux系统: Debian操作系统常用命令指南

高效管理你的Linux系统: Debian操作系统常用命令指南

《高效管理你的Linux系统:Debian操作系统常用命令指南》在Debian操作系统中，了解和掌握常用命令对于提高工作效率和系统管理至关重要，本文将详细介绍Debian的常用命令，帮助读者更好地使... Debian是一个流行的linux发行版，它以其稳定性、强大的软件包管理和丰富的社区资源而闻名。在使用
阅读更多...
Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单

Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单

《Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单》：本文主要介绍Springboot的ThreadPoolTaskScheduler线... 目录ThreadPoolTaskScheduler线程池实现15分钟不操作自动取消订单概要1，创建订单后
阅读更多...
SpringBoot使用minio进行文件管理的流程步骤

SpringBoot使用minio进行文件管理的流程步骤

《SpringBoot使用minio进行文件管理的流程步骤》MinIO是一个高性能的对象存储系统,兼容AmazonS3API,该软件设计用于处理非结构化数据,如图片、视频、日志文件以及备份数据等,本文... 目录一、拉取minio镜像二、创建配置文件和上传文件的目录三、启动容器四、浏览器登录 minio五、
阅读更多...
C语言线程池的常见实现方式详解

C语言线程池的常见实现方式详解

《C语言线程池的常见实现方式详解》本文介绍了如何使用C语言实现一个基本的线程池,线程池的实现包括工作线程、任务队列、任务调度、线程池的初始化、任务添加、销毁等步骤,感兴趣的朋友跟随小编一起看看吧... 目录1. 线程池的基本结构2. 线程池的实现步骤3. 线程池的核心数据结构4. 线程池的详细实现4.1 初
阅读更多...
IDEA中的Kafka管理神器详解

IDEA中的Kafka管理神器详解

《IDEA中的Kafka管理神器详解》这款基于IDEA插件实现的Kafka管理工具,能够在本地IDE环境中直接运行,简化了设置流程,为开发者提供了更加紧密集成、高效且直观的Kafka操作体验... 目录免安装：IDEA中的Kafka管理神器！简介安装必要的插件创建 Kafka 连接第一步：创建连接第二步：选
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2