杭州网站提升排名,如何制定会员营销方案,上海 建站,虎丘网站建设目录
23.线程池 23.1 什么是线程池 23.2 为什么需要线程池 23.3 线程池的应用场景 23.4 实现一个简单的线程池 23.4.1 RAII风格信号锁 23.4.2 线程的封装 23.4.3 日志打印 22.4.4 定义队列中存放Task类任务 23.4.5 线程池的实现(懒汉模式) 为什么线程池中需要有互斥锁和条件变…目录
23.线程池 23.1 什么是线程池 23.2 为什么需要线程池 23.3 线程池的应用场景 23.4 实现一个简单的线程池 23.4.1 RAII风格信号锁 23.4.2 线程的封装 23.4.3 日志打印 22.4.4 定义队列中存放Task类任务 23.4.5 线程池的实现(懒汉模式) 为什么线程池中需要有互斥锁和条件变量? 为什么线程池中的线程执行例程需要设置为静态方法 为什么私有化构造函数?(单例模式) 23.4.6 单例模式 饿汉方式实现单例模式 懒汉方式实现单例模式 23.4.7 主程序实现 22.4.8 程序运行效果 23.线程池 23.1 什么是线程池
线程池Thread Pool是一种管理和重用线程的机制,在多线程编程中创建和销毁线程是一项开销较大的操作因为涉及到操作系统的资源管理和线程上下文切换的成本。线程池通过预先创建一组线程并维护它们可以避免频繁地创建和销毁线程从而降低了系统的开销。 线程池中的多个已经创建好的线程负责从任务队列当中拿任务并将拿到的任务进行处理。线程池对外提供一个Push接口用于让外部将任务Push到任务队列当中 23.2 为什么需要线程池
举个例子 想象你是一个公司的HR经理有许多员工需要执行不同的任务。每当有任务到来时你要做的是从员工池中选择一个空闲的员工来执行任务。如果每次任务到来时都去招聘新员工创建新线程等任务完成后再解雇他们销毁线程那么你将花费大量的时间和精力来管理这个过程而且频繁的招聘和解雇会带来一定的成本和效率损失。相反如果你有一个固定数量的员工池线程池这些员工已经预先准备好并且处于待命状态。当任务到来时你只需要从员工池中选择一个空闲的员工分配任务给他完成后他会继续留在池中等待下一个任务的到来。这样可以避免频繁的人员调配线程的创建和销毁节省了时间和成本提高了效率。因此线程池就像是一个员工池能够有效地管理和利用资源提高系统的性能和效率。 当需要处理大量的并发任务时使用线程池能够提高系统的效率和性能 :
减少线程创建开销线程池通过维护一定数量的预创建线程避免了频繁创建和销毁线程所带来的系统资源消耗。这样一来系统能够更有效地利用可用的处理器资源和内存提高整体的资源利用率。任务队列管理线程池通常配备了任务队列用于存储需要执行的任务。通过合理管理任务队列可以实现任务的调度和分配保证任务的有序执行提高系统的整体性能。提高代码可维护性使用线程池能够将线程管理的逻辑与业务逻辑分离使代码更加清晰简洁降低了系统的复杂度提高了代码的可维护性和可读性。控制并发度线程池允许开发者限制同时执行的线程数量从而避免系统资源被过度占用防止系统负载过重提高系统的稳定性和可靠性。 23.3 线程池的应用场景
线程池适合应对那些需要异步执行、并发处理的任务。适合处理以下类型的任务 短时任务Short-lived Tasks短时任务是指执行时间较短的任务通常是一些简单的计算、请求响应等操作。线程池可以有效地管理和复用线程避免频繁创建和销毁线程的开销从而更快地处理这类任务。事件驱动任务Event-driven Tasks事件驱动任务是指需要响应事件触发的任务如GUI应用程序中的事件处理、网络服务器中的请求处理等。线程池可以用来处理这些事件保持系统的响应速度和并发处理能力。IO密集型任务IO-bound TasksIO密集型任务是指任务主要涉及到IO操作如文件读写、网络请求等。由于IO操作通常会阻塞线程使用线程池可以将阻塞操作委托给后台线程处理从而提高系统的并发处理能力。 这使得线程池在许多不同的应用场景中都能发挥重要作用。以下是一些常见的应用场景 Web 开发在 Web 开发中常常需要处理大量的并发请求如处理用户请求、数据库查询等。通过使用线程池可以提高服务器的并发处理能力减少请求的响应时间。 后台任务处理在后台任务处理中如批量数据处理、定时任务调度等使用线程池可以更有效地管理和调度任务提高系统的性能和效率。 图像处理和计算密集型任务在图像处理、视频编解码、科学计算等领域常常需要进行大量的计算密集型任务。使用线程池可以并行地处理这些任务提高处理速度和效率。 数据库连接管理在数据库访问中通过使用线程池管理数据库连接可以减少数据库连接的创建和销毁开销提高数据库访问的性能和效率。 并行编程框架在并行编程框架中如 Java 的 Fork/Join 框架、Python 的 concurrent.futures 模块等线程池常常被用来并行地执行任务以提高程序的性能和扩展性。 23.4 实现一个简单的线程池
23.4.1 RAII风格信号锁
我们可以定义了一个 lockGuard 类采用 RAII资源获取即初始化方式对互斥锁进行加锁和解锁确保在作用域结束时自动释放锁。
这里我们创建一个名为lock.hpp的文件来定义lockGuard类
#pragma once
#include iostream
#include pthread.h
class lockGuard
{
public:lockGuard(pthread_mutex_t *mtx) : mtx_(mtx){pthread_mutex_lock(mtx_);}~lockGuard(){pthread_mutex_unlock(mtx_);}private:pthread_mutex_t *mtx_; // 指向要管理的互斥锁的指针
};在lockGuard类的构造函数中首先通过传入的pthread_mutex_t类型的指针初始化mtx_成员变量即指向要管理的互斥锁。然后调用pthread_mutex_lock函数对该互斥锁进行加锁操作。
在lockGuard类的析构函数中调用pthread_mutex_unlock函数对互斥锁进行解锁操作。由于该析构函数在对象生命周期结束时自动调用因此实现了互斥锁的自动释放。这样在使用lockGuard对象时只需要在作用域中创建该对象当对象离开作用域时析构函数会自动调用从而释放互斥锁确保了互斥锁的安全管理。
23.4.2 线程的封装
这里我们创建一个名为Thread.hpp的文件来定义Thread类,实现对其的封装.
#pragma once
#include iostream
#include string
#include functional
#include cstdio
#include pthread.h // 使用 pthread 相关函数需要包含 pthread.h 头文件// 定义一个函数指针类型用于作为线程的回调函数
typedef void *(*fun_t)(void *);// 线程的数据结构用于传递给线程执行函数
class ThreadData
{
public:void *args_; // 线程执行函数的参数std::string name_; // 线程的名称
};// 线程类用于创建和管理线程
class Thread
{
public:// 构造函数初始化线程对象Thread(int num, fun_t callback, void *args) : func_(callback){char nameBuffer[64];snprintf(nameBuffer, sizeof nameBuffer, Thread-%d, num);name_ nameBuffer;tdata_.args_ args;tdata_.name_ name_;}// 启动线程void start(){pthread_create(tid_, nullptr, func_, (void *)tdata_);}// 等待线程结束void join(){pthread_join(tid_, nullptr);}// 返回线程的名称std::string name(){return name_;}// 析构函数~Thread(){}private:std::string name_; // 线程名称fun_t func_; // 线程执行的回调函数ThreadData tdata_; // 线程的数据pthread_t tid_; // 线程ID
};23.4.3 日志打印
这里我们创建一个名为Log.hpp的文件来实现日志打印保存
#pragma once
#include iostream
#include cstdio
#include cstdarg
#include ctime
#include string
#define DEBUG 0
#define NORMAL 1
#define WARNING 2
#define ERROR 3
#define FATAL 4
const char *gLevelMap[] {DEBUG,NORMAL,WARNING,ERROR,FATAL
};
#define LOGFILE ./threadpool.log
void logMessage(int level, const char *format, ...)
{
#ifndef DEBUG_SHOWif(level DEBUG) return;
#endifchar stdBuffer[1024]; //标准部分time_t timestamp time(nullptr);snprintf(stdBuffer, sizeof stdBuffer, [%s] [%ld] , gLevelMap[level], timestamp);char logBuffer[1024]; //自定义部分va_list args;va_start(args, format);vsnprintf(logBuffer, sizeof logBuffer, format, args);va_end(args);FILE *fp fopen(LOGFILE, a);fprintf(fp, %s%s\n, stdBuffer, logBuffer);fclose(fp);
}
这段代码定义了一个函数 logMessage用于记录日志到一个指定的文件中。日志级别被定义为 DEBUG、NORMAL、WARNING、ERROR 和 FATAL不同级别的日志会按照不同的格式输出到日志文件中。日志文件名被定义为 ./threadpool.log 。编译时如果未显示定义 -DEBUG_SHOW ,则不会将DEBUG 级别的日志打印进文件
va_list args;
va_start(args, format);
vsnprintf(logBuffer, sizeof logBuffer, format, args);
va_end(args);
这段代码片段使用了 C 语言的可变参数函数和相关的宏来格式化日志消息。 va_list 是一个类型用于声明一个指向参数列表的变量。在这里args 是一个 va_list 类型的变量用于存储可变参数列表。va_start 宏初始化 args 变量使其指向参数列表中的第一个可变参数。在这里args 变量将指向传递给 logMessage 函数的第一个可变参数。vsnprintf 函数将可变参数列表按照指定的格式 format 进行格式化并将结果存储到 logBuffer 中。它的功能类似于 sprintf 函数但是可以处理可变参数列表。va_end 宏清理 args 变量结束对可变参数列表的访问。在这里它通知编译器该可变参数列表已经处理完毕可以进行清理工作。 22.4.4 定义队列中存放Task类任务
这里我们创建一个名为Task.hpp的文件来定义Task类
#pragma once
#include iostream
#include string
#include functional
#include Log.hpp
typedef std::functionint(int, int) func_t;
class Task
{
public:Task(){}Task(int x, int y, func_t func):x_(x), y_(y), func_(func){}void operator ()(const std::string name){logMessage(WARNING, %s处理完成: %d%d%d | %s | %d,name.c_str(), x_, y_, func_(x_, y_), __FILE__, __LINE__);}
public:int x_;int y_;func_t func_;
};
23.4.5 线程池的实现(懒汉模式)
我们需要实现Push接口向任务队列里面放任务,又要Pop任务给已经创建好的线程,其本质就是一个生产者-消费者模型 这里我们创建一个名为ThreadPool.hpp的文件来定义ThreadPool类
#pragma once
#include iostream
#include vector
#include string
#include queue
#include unistd.h
#include Thread.hpp // 导入线程类头文件
#include Lock.hpp // 导入锁相关的头文件
#include Log.hpp // 导入日志相关的头文件const int g_thread_num 3; // 默认线程池中线程数量// 线程池类模板
template class T
class ThreadPool
{
public:pthread_mutex_t *getMutex() // 获取互斥锁指针{return lock;}bool isEmpty() // 判断任务队列是否为空{return task_queue_.empty();}void waitCond() // 等待条件变量{pthread_cond_wait(cond, lock);}T getTask() // 获取任务{T t task_queue_.front(); // 获取队首任务task_queue_.pop(); // 移除队首任务return t;}
private:ThreadPool(int thread_num g_thread_num) : num_(thread_num) // 构造函数{pthread_mutex_init(lock, nullptr); // 初始化互斥锁pthread_cond_init(cond, nullptr); // 初始化条件变量for (int i 1; i num_; i){threads_.push_back(new Thread(i, routine, this)); // 创建线程对象并加入线程池}}ThreadPool(const ThreadPoolT other) delete; // 禁止拷贝构造函数const ThreadPoolT operator(const ThreadPoolT other) delete; // 禁止赋值运算符public:static ThreadPoolT *getThreadPool(int num g_thread_num) // 获取线程池单例{if (nullptr thread_ptr) // 如果线程池指针为空{lockGuard lockguard(mutex); // 创建锁保护if (nullptr thread_ptr){thread_ptr new ThreadPoolT(num); // 创建线程池对象}}return thread_ptr; // 返回线程池指针}void Threadinit() // 初始化线程池中的线程{for (auto iter : threads_) // 遍历线程池中的线程对象{iter-start(); // 启动线程logMessage(NORMAL, %s %s, iter-name().c_str(), 启动成功); // 记录日志}}static void *routine(void *args) // 线程执行函数{ThreadData *td (ThreadData *)args; // 获取线程数据ThreadPoolT *tp (ThreadPoolT *)td-args_; // 获取线程池指针while (true){T task;{lockGuard lockguard(tp-getMutex()); // 加锁while (tp-isEmpty()) // 如果任务队列为空tp-waitCond(); // 等待条件变量task tp-getTask(); // 获取任务}task(td-name_); // 执行任务}}void pushTask(const T task) // 添加任务到任务队列{lockGuard lockguard(lock); // 加锁task_queue_.push(task); // 添加任务到队列pthread_cond_signal(cond); // 发送信号唤醒线程}~ThreadPool() // 析构函数{for (auto iter : threads_) // 遍历线程池中的线程对象{iter-join(); // 等待线程结束delete iter; // 删除线程对象}pthread_mutex_destroy(lock); // 销毁互斥锁pthread_cond_destroy(cond); // 销毁条件变量}
private:std::vectorThread * threads_; // 线程对象的容器int num_; // 线程数量std::queueT task_queue_; // 任务队列static ThreadPoolT *thread_ptr; // 线程池指针static pthread_mutex_t mutex; // 互斥锁pthread_mutex_t lock; // 互斥锁pthread_cond_t cond; // 条件变量
};// 初始化线程池指针为nullptr
template typename T
ThreadPoolT *ThreadPoolT::thread_ptr nullptr;// 初始化互斥锁
template typename T
pthread_mutex_t ThreadPoolT::mutex PTHREAD_MUTEX_INITIALIZER;为什么线程池中需要有互斥锁和条件变量? 线程池中的任务队列是会被多个执行流同时访问的临界资源因此我们需要引入互斥锁对任务队列进行保护。线程池当中的线程要从任务队列里拿任务前提条件是任务队列中必须要有任务因此线程池当中的线程在拿任务之前需要先判断任务队列当中是否有任务若此时任务队列为空那么该线程应该进行等待直到任务队列中有任务时再将其唤醒因此我们需要引入条件变量。当外部线程向任务队列中Push一个任务后此时可能有线程正处于等待状态因此在新增任务后需要唤醒在条件变量下等待的线程。 注意: 使用 while 进行条件判断在使用条件变量等待线程唤醒时应该使用 while 循环而不是 if 语句进行条件判断。这是因为唤醒线程可能会出现伪唤醒spurious wakeup的情况或者是其他导致唤醒的条件。使用 while 循环可以确保线程在被唤醒后重新检查条件以防止出现虚假唤醒的情况。 避免惊群效应确实使用 pthread_cond_broadcast 可能会导致惊群效应即一次性唤醒大量线程而实际上只有一个线程能够获取到任务。为了避免这种情况最好使用 pthread_cond_signal 函数唤醒一个等待的线程即可这样可以减少不必要的竞争和资源浪费。 在解锁之后处理任务正确的做法是在拿到任务后先解锁然后再处理任务。这样可以确保其他线程能够及时获取到任务并开始处理而不会因为一个线程长时间持有锁而导致其他线程无法执行的情况。将任务处理过程放到解锁之前可能会导致资源的浪费和效率降低。 避免串行执行如果将处理任务的过程放到临界区当中那么当某一线程从任务队列拿到任务后其他线程还需要等待该线程将任务处理完后才有机会进入临界区。此时虽然是线程池但最终我们可能并没有让多线程并行的执行起来。 为什么线程池中的线程执行例程需要设置为静态方法 当使用pthread_create函数创建线程时需要提供一个C风格的函数指针作为线程的执行例程。然而在C中类的成员函数会隐式地接收一个指向该类对象的指针作为第一个参数即this指针。因此如果直接将类的成员函数作为线程的执行例程会导致函数签名不匹配无法通过编译。
为了解决这个问题一种常见的方法是将线程执行例程设置为静态方法。静态方法不会隐式接收this指针作为参数而是直接通过类名调用。这样就可以在pthread_create函数中传递一个静态成员函数的指针作为线程的执行例程。同时静态方法也具有与类对象实例无关的特性因此可以被多个线程共享。
然而静态方法无法直接访问非静态成员函数和变量。如果在执行例程中需要访问类的非静态成员函数或变量(比如Pop)可以通过在创建线程时向执行例程传递类对象的指针来实现。这样在执行例程中就可以使用该指针来调用类的非静态成员函数或访问非静态成员变量。 为什么私有化构造函数?(单例模式) 私有化构造函数的主要原因是确保线程池类 ThreadPoolT 的单例模式。通过私有化构造函数外部无法直接创建新的线程池实例而是必须通过静态方法 getThreadPool() 来获取线程池的唯一实例。
23.4.6 单例模式
单例Singleton模式是一种常用的软件设计模式。在它的核心结构中只包含一个被称为单例的特殊类。通过单例模式可以保证系统中应用该模式的类一个类只有一个实例。即一个类只有一个对象实例 使用场景 语义上只需要一个该对象内部存在大量的空间保存了大量的数据如果允许该对象存在多份或者允许发生各种拷贝内存中存在冗余数据 一般Singleton模式通常有三种形式: 饿汉方式吃完饭, 立刻洗碗, 这种就是饿汉方式. 因为下一顿吃的时候可以立刻拿着碗就能吃饭。懒汉方式吃完饭, 先把碗放下, 然后下一顿饭用到这个碗了再洗碗, 就是懒汉方式。 懒汉方式最核心的思想是 延时加载。例如我们之前所学过的写时拷贝从而能够优化服务器的启动速度。 饿汉方式实现单例模式
template typename T
class Singleton
{
private:static SingletonT data;//饿汉模式在加载的时候对象就已经存在了
public: static SingletonT* GetInstance() { return data; }
};
该模式能简单快速的创建一个单例对象而且是线程安全的(只在类加载时才会初始化以后都不会)。但它有一个缺点就是不管你要不要都会直接创建一个对象会消耗一定的性能(当然很小很小几乎可以忽略不计所以这种模式在很多场合十分常用而且十分简单)
懒汉方式实现单例模式
该模式只在你需要对象时才会生成单例对象(比如调用GetInstance方法)
template typename T
class Singleton
{
private:static SingletonT* inst; //懒汉式单例只有在调用GetInstance时才会实例化一个单例对象
public: static SingletonT* GetInstance() { if (inst NULL) { inst new SingletonT(); }return inst; }
};
看上去这段代码没什么明显问题但它不是线程安全的。假设当前有多个线程同时调用GetInstance方法由于当前还没有对象生成那么就会由多个线程创建多个对象。
// 懒汉模式, 线程安全
template typename T
class Singleton
{
private: static SingletonT* inst; static std::mutex lock;
public: static T* GetInstance() { if (inst NULL) // 双重判定空指针, 降低锁冲突的概率, 提高性能 { lock.lock(); // 使用互斥锁, 保证多线程情况下也只调用一次 newif (inst NULL) { inst new T(); }lock.unlock(); }return inst;}
};
这种形式是在懒汉方式的基础上增加的当多个线程调用GetInstance方法时此时类中没有对象那么多个线程就会来到锁的位置竞争锁。必然只能有一个线程竞争锁成功此时再次判断有没有对象被创建就是inst指针如果没有就会new一个对象如果有就会解锁并返回已有的对象总的来说这样的形式使得多个线程调用GetInstance方法时无论成功与否都会有返回值
23.4.7 主程序实现
#includeLog.hpp
#includeThreadPool.hpp
#includeTask.hpp
#include ctime
#include cstdlib
#include iostream
#include unistd.h
int main()
{srand((unsigned long)time(nullptr) ^ getpid());ThreadPoolTask::getThreadPool()-Threadinit();while(true){int x rand()%100 1;usleep(7721);int y rand()%30 1;Task t(x, y, [](int x, int y)-int{return x y;});logMessage(DEBUG, 制作任务完成: %d%d?, x, y);logMessage(DEBUG, 制作任务完成: %d%d?, x, y);logMessage(DEBUG, 制作任务完成: %d%d?, x, y);logMessage(DEBUG, 制作任务完成: %d%d?, x, y);ThreadPoolTask::getThreadPool()-pushTask(t);sleep(1);}return 0;
}22.4.8 程序运行效果
程序运行后,我们打开路径下自动新建得threadpool.log文件,可以看到程序自动打印的日志
