从头开始：构建一个基于C/C++的线程池

🕗 发布于 2024-09-25 05:56 c语言 c++ 网络

手搓线程池

线程池工作原理和实现

线程池工作原理和实现

线程池工作原理

线程池（Thread Pool）是一种预先创建和管理一组工作线程的技术，用来优化并发任务的执行。通过复用这些线程来执行多个任务，线程池可以减少线程创建和销毁的开销，提高系统的性能和响应速度。

1. 线程池的基本组成：

任务队列（Task Queue）：存放待执行任务的队列。当有新的任务提交时，它会进入任务队列，等待可用线程来执行。
工作线程（Worker Threads）：线程池中的一组线程，用来处理任务队列中的任务。线程池启动时会创建一定数量的工作线程。
线程池管理器（Thread Pool Manager）：负责管理线程池的大小、任务分配、线程的创建和销毁。

2. 线程池的基本执行流程：

提交任务：用户向线程池提交任务，通常是实现了 Runnable 或 Callable 接口的对象。任务被放入任务队列中。
任务分配：线程池中的某个空闲工作线程会从任务队列中取出一个任务，并开始执行。任务队列是线程安全的，确保多个线程可以同时取任务而不发生冲突。
任务执行：工作线程运行并执行任务中的代码。线程池中的工作线程是循环的，每个线程在完成一个任务后会继续取下一个任务。
复用线程：任务完成后，线程不会被销毁，而是回到线程池中成为“空闲线程”，等待下一个任务。
动态调整线程数量：如果任务数量激增，线程池可以根据策略（如扩展线程池的大小）创建新的线程来处理更多的任务；如果任务减少，线程池也可能会销毁一些线程以节省资源。

3. 线程池的核心参数：

核心线程数（corePoolSize）：线程池中保持活跃的核心线程数量，即使线程处于空闲状态也不会被回收。
最大线程数（maximumPoolSize）：线程池中允许的最大线程数量。当任务数量超过核心线程数时，线程池会根据需求创建更多的线程，直到达到最大线程数。
任务队列（workQueue）：用于存放等待执行的任务。如果当前所有线程都在忙碌，新的任务会被放入任务队列中。
线程存活时间（keepAliveTime）：线程池中超过核心线程数的空闲线程在等待新任务的最长等待时间，超过这个时间后将被销毁。
线程工厂（ThreadFactory）：用于创建新线程，方便自定义线程的属性，如线程的名称、优先级等。
拒绝策略（RejectedExecutionHandler）：当任务过多而无法处理时（任务队列满了且线程池的线程数已达到上限），线程池会执行拒绝策略，常见策略包括丢弃任务、抛出异常、或由调用者线程直接执行。

4. 线程池的生命周期：

运行状态（RUNNING）：线程池处于运行状态，可以接受任务并处理任务。
关闭状态（SHUTDOWN）：线程池不再接受新任务，但会继续处理已经提交的任务。
停止状态（STOP）：线程池不再接受任务，且会中断正在执行的任务。
终止状态（TERMINATED）：所有任务执行完毕，线程池中的线程全部销毁，线程池彻底关闭。

5. 线程池的执行策略：

先使用核心线程：线程池优先利用核心线程来处理任务，只有在核心线程全部繁忙的情况下，才会将任务放入任务队列。
任务队列满了，创建新线程：如果任务队列已满且所有核心线程都在工作，线程池会创建新的线程，直到达到 maximumPoolSize。
拒绝任务：如果任务队列满了，且线程池中的线程数量已经达到 maximumPoolSize，根据配置的拒绝策略处理新任务。

基于C语言的线程池设计与实现

任务队列

typedef struct Task
{
    void (*function) (void* arg); // void*是一个泛型，能够接收各种各样的数据类型
    void* arg;
}Task;

线程池定义

struct ThreadPool
{
    // 任务队列
    Task* taskQ; // 队列数组
    int queueCapacity;  // 容量
    int queueSize;      // 当前任务个数
    int queueFront;     // 队头，取数据
    int queueRear;      // 队尾，放数据

    pthread_t managerID;    // 管理者线程ID
    pthread_t *threadIDs;   // 工作的线程ID
    int minNum;             // 最小的线程数
    int maxNum;             // 最大的线程数
    int busyNum;            // 工作的线程个数
    int liveNum;            // 存活的线程个数
    int exitNum;            // 要杀死的线程个数
    pthread_mutex_t mutexpool;  // 锁住整个线程池
    pthread_mutex_t mutexBusy;  // 锁住busyNum变量
    pthread_cond_t notFull;     // 任务队列是否满了
    pthread_cond_t notEmpty;    // 任务队列是否空了

    int shutdown;               // 是否销毁线程池，销毁为1，不销毁为0
};

头文件声明

#ifndef _THREADPOOL_H
#define _THREADPOOL_H

typedef struct ThreadPool ThreadPool;
// 创建线程池并初始化
ThreadPool* threadPoolCreate(int min, int max, int queueSize);

// 销毁线程池
int threadPoolDestory(ThreadPool* pool);

// 给线程池添加任务
void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg);

// 获取线程池中工作的线程的个数
int threadPoolBusyNum(ThreadPool* pool);

// 获取线程池中活着的线程的个数
int threadPoolAliveNum(ThreadPool* pool);

void* worker(void* arg);

void* manager(void* arg);

void threadExit(ThreadPool* pool);

#endif

源文件定义

#include "threadpool.h"
#include<pthread.h>
#include<stdlib.h>
#include<string.h>
#include<unistd.h>
#include<stdio.h>

const int NUMBER = 2;

// 任务结构体
typedef struct Task
{   
    // void*是一个泛型，能够接收各种各样的数据类型
    void (*function) (void* arg); // 函数指针
    void* arg;
}Task;


// 线程池结构体
struct ThreadPool
{
    // 任务队列
    Task* taskQ; // 队列数组
    int queueCapacity;  // 容量
    int queueSize;      // 当前任务个数
    int queueFront;     // 队头，取数据
    int queueRear;      // 队尾，放数据

    pthread_t managerID;    // 管理者线程ID
    pthread_t *threadIDs;   // 工作的线程ID
    int minNum;             // 最小的线程数
    int maxNum;             // 最大的线程数
    int busyNum;            // 工作的线程个数
    int liveNum;            // 存活的线程个数
    int exitNum;            // 要杀死的线程个数
    pthread_mutex_t mutexpool;  // 锁住整个线程池
    pthread_mutex_t mutexBusy;  // 锁住busyNum变量
    pthread_cond_t notFull;     // 任务队列是否满了，用于阻塞生产者
    pthread_cond_t notEmpty;    // 任务队列是否空了，用于阻塞消费者

    int shutdown;               // 是否销毁线程池，销毁为1，不销毁为0
};

ThreadPool* threadPoolCreate(int min, int max, int queueSize) {
    ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
    do {
        if(pool == NULL) {
            printf("malloc threadpool fail....\n");
            break;
        }
        pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * max);
        if(pool->threadIDs == NULL) {
            printf("malloc threadIDs fail....\n");
            break;
        }
        memset(pool->threadIDs, 0, sizeof(pthread_t) * max);
        pool->minNum = min;
        pool->maxNum = max;
        pool->busyNum = 0;
        pool->liveNum = min; // 和最小个数相等
        pool->exitNum = 0;

        if(pthread_mutex_init(&pool->mutexpool, NULL) != 0 ||
        pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
        pthread_cond_init(&pool->notEmpty, NULL) != 0 ||
        pthread_cond_init(&pool->notFull, NULL) != 0
        ) {
            printf("mutex or condif init fail....\n");
            break;
        }

        // 任务队列
        pool->taskQ = (Task*)malloc(sizeof(Task) * queueSize);
        pool->queueCapacity = queueSize;
        pool->queueSize = 0;
        pool->queueFront = 0;
        pool->queueRear = 0;

        pool->shutdown = 0;

        // 创建管理者线程和工作者线程
        pthread_create(&pool->managerID, NULL, manager, pool); // 第三个参数为管理者线程的任务函数
        for(int i = 0; i < min; i++) {
            pthread_create(&pool->threadIDs[i], NULL, worker, pool); // 第三个参数为工作的线程的任务函数
        }
        return pool;
    } while(0);

    // 释放资源
    if(pool->threadIDs) free(pool->threadIDs);
    if(pool->taskQ) free(pool->taskQ);
    if(pool) free(pool);

    return NULL;
}

int threadPoolDestory(ThreadPool* pool) {
    if(pool == NULL) {
        return -1;
    }

    // 关闭线程池
    pool->shutdown = 1;
    // 阻塞回收管理者线程
    pthread_join(pool->managerID, NULL);
    // 唤醒阻塞的消费者线程
    for(int i = 0; i < pool->liveNum; i++) {
        pthread_cond_signal(&pool->notEmpty);
    }
    // 释放堆内存
    if(pool->taskQ) {
        free(pool->taskQ);
    }
    if(pool->threadIDs) {
        free(pool->threadIDs);
    }
    
    pthread_mutex_destroy(&pool->mutexpool);
    pthread_mutex_destroy(&pool->mutexBusy);
    pthread_cond_destroy(&pool->notEmpty);
    pthread_cond_destroy(&pool->notFull);
    free(pool);
    pool = NULL;
    return 0;
}

void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg) {
    pthread_mutex_lock(&pool->mutexpool);
    while(pool->queueSize == pool->queueCapacity && !pool->shutdown) {
        // 阻塞生产者线程
        pthread_cond_wait(&pool->notFull, &pool->mutexpool);
    }
    if(pool->shutdown) {
        pthread_mutex_unlock(&pool->mutexpool);
        return;
    }
    // 添加任务
    pool->taskQ[pool->queueRear].function = func;
    pool->taskQ[pool->queueRear].arg = arg;
    pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity;
    pool->queueSize++;

    pthread_cond_signal(&pool->notEmpty); // 唤醒阻塞在条件变量里的工作的线程，即生产者生产后唤醒消费者
    pthread_mutex_unlock(&pool->mutexpool);
}

int threadPoolBusyNum(ThreadPool* pool) {
    pthread_mutex_lock(&pool->mutexBusy);
    int busyNum = pool->busyNum;
    pthread_mutex_unlock(&pool->mutexBusy);
    return busyNum;
}

int threadPoolAliveNum(ThreadPool* pool) {
    pthread_mutex_lock(&pool->mutexpool);
    int liveNum = pool->liveNum;
    pthread_mutex_unlock(&pool->mutexpool);
    return liveNum;
}

void* worker(void* arg) {
    ThreadPool* pool = (ThreadPool*)arg;
    while(1) {
        pthread_mutex_lock(&pool->mutexpool); // 访问线程池之前加锁
        // 当前任务队列是否为空
        while(pool->queueSize == 0 && !pool->shutdown) { // 任务队列为空并且线程池没有被关闭
            // 阻塞工作线程
            pthread_cond_wait(&pool->notEmpty, &pool->mutexpool);

            // 判断是不是要销毁线程
            if(pool->exitNum > 0) {
                pool->exitNum--;
                if(pool->liveNum > pool->minNum) {
                    pool->liveNum--;
                    pthread_mutex_unlock(&pool->mutexpool); // 阻塞的时候已经获得锁，已经锁上，如果不解开就会死锁
                    threadExit(pool);
                }
            }
        }

        // 判断线程池是否被关闭了
        if(pool->shutdown) {
            pthread_mutex_unlock(&pool->mutexpool); // 避免死锁
            threadExit(pool);
        }

        // 从任务队列中取出一个任务
        Task task;
        task.function = pool->taskQ[pool->queueFront].function;
        task.arg = pool->taskQ[pool->queueFront].arg;
        // 移动头结点
        pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity; // 循环队列
        pool->queueSize--;

        pthread_cond_signal(&pool->notFull); // 消费者消费完产品后唤醒生产者
        pthread_mutex_unlock(&pool->mutexpool); // 用完之后解锁

        printf("thread %ld start working...\n", pthread_self());
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum++;
        pthread_mutex_unlock(&pool->mutexBusy);
        task.function(task.arg);
        free(task.arg);
        task.arg = NULL;

        printf("thread %ld end working...\n", pthread_self());
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum--;
        pthread_mutex_unlock(&pool->mutexBusy);
    }
    return NULL;
}

void* manager(void* arg) {
    ThreadPool* pool = (ThreadPool*)arg;
    while(!pool->shutdown) {
        // 每隔三秒钟检测一次
        sleep(3);

        // 取出线程池中任务的数量和当前线程的数量
        pthread_mutex_lock(&pool->mutexpool);
        int queueSize = pool->queueSize;
        int liveNumber = pool->liveNum;
        pthread_mutex_unlock(&pool->mutexpool);

        // 取出忙的线程的数量
        pthread_mutex_lock(&pool->mutexBusy);
        int busyNumber = pool->busyNum;
        pthread_mutex_unlock(&pool->mutexBusy);

        // 添加线程
        // 任务的个数 大于 存活的线程个数 && 存活的线程数 小于 最大线程数
        if(queueSize > liveNumber && liveNumber < pool->maxNum) {
            pthread_mutex_lock(&pool->mutexpool);
            int count = 0;
            for(int i = 0; i < pool->maxNum && count < NUMBER && liveNumber < pool->maxNum; i++) {
                if(pool->threadIDs[i] == 0) {
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool);
                    count++;
                    pool->liveNum++;
                }
            }
            pthread_mutex_unlock(&pool->mutexpool);
        }

        // 销毁线程
        // 忙的线程 * 2 小于 存活的线程数 && 存活的线程 大于 最小线程数
        // 之所以不用对pool->minNum加锁，是因为这个值是固定的，只需要读不需要写
        if(busyNumber * 2 < liveNumber && liveNumber > pool->minNum) {
            pthread_mutex_lock(&pool->mutexpool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutexpool);
            // 让工作的线程自杀
            for(int i = 0; i < NUMBER; i++) {
                pthread_cond_signal(&pool->notEmpty);
            }
        }
    }
}

void threadExit(ThreadPool* pool) {
    pthread_t tid = pthread_self(); // 获得线程自身的ID。
    for(int i = 0; i < pool->maxNum; i++) {
        if(pool->threadIDs[i] == tid) { // tid对应的线程要退出了
            pool->threadIDs[i] = 0;
            printf("threadExit() called, %ld exiting...\n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}

测试代码

#include<stdio.h>
#include "threadpool.h"
#include<pthread.h>
#include<unistd.h>
#include<stdlib.h>
#include<string.h>

void taskFunc(void* arg) {
    int num = *(int*)arg;
    printf("thread %ld is working, number = %d\n", pthread_self(), num);
    sleep(1);
}

int main() {
    // 创建线程池
    ThreadPool* pool = threadPoolCreate(3, 10, 100);
    for(int i = 0; i < 100; i++) {
        int* num = (int*)malloc(sizeof(int));
        *num = i + 100;
        threadPoolAdd(pool, taskFunc, num);
    }

    sleep(30); // 等待子线程把任务处理完毕

    threadPoolDestory(pool);
    return 0;
}

基于C++的线程池设计与实现

因为在C++中，delete task.arg时候，由于delete void*类型是有危险的，因为void*指针只占四个字节，因此有可能不能全部地被释放，为了知道在程序中void*实际上是什么类型，因此在C++中可以使用模板来解决这一问题，因为模板可以传递类型。

任务队列声明

#pragma once
#include<queue>
#include<pthread.h>

using callback = void (*) (void* arg);

// 任务结构体
template <typename T>
struct Task
{   
    Task<T>() {
        function = nullptr;
        arg = nullptr;
    }
    Task<T>(callback f, void* arg) {
        this->arg = (T*)arg;
        function = f;
    }
    callback function; // 函数指针
    T* arg;
};

template <typename T>
class TaskQueue
{
private:
    std::queue <Task<T>> m_taskQ;
    pthread_mutex_t m_mutex; // 互斥锁
public:
    TaskQueue(/* args */);
    ~TaskQueue();

    // 添加任务
    void addTask(Task<T> task);
    void addTask(callback f, void* arg);
    // 取出一个任务
    Task<T> takeTask();
    // 获取当前任务的个数
    inline size_t taskNumber() { // 没有if判断什么的，比较简单的直接写成内联函数比较好
        return m_taskQ.size();
    }
};

任务队列定义

#include "TaskQueue.h"


template <typename T>
TaskQueue<T>::TaskQueue(/* args */)
{
    pthread_mutex_init(&m_mutex, NULL);
}

template <typename T>
TaskQueue<T>::~TaskQueue()
{
    pthread_mutex_destroy(&m_mutex);
}

template <typename T>
void TaskQueue<T>::addTask(Task<T> task)
{
    pthread_mutex_lock(&m_mutex);
    m_taskQ.push(task);
    pthread_mutex_unlock(&m_mutex);
}

template <typename T>
void TaskQueue<T>::addTask(callback f, void* arg)
{
    pthread_mutex_lock(&m_mutex);
    m_taskQ.push(Task<T>(f, arg));
    pthread_mutex_unlock(&m_mutex);
}

template <typename T>
Task<T> TaskQueue<T>::takeTask() 
{
    Task<T> task;
    pthread_mutex_lock(&m_mutex);
    if(!m_taskQ.empty()) {
        task = m_taskQ.front();
        m_taskQ.pop();
    }
    pthread_mutex_unlock(&m_mutex);
    return task;
}

线程池声明

#pragma once
#include "TaskQueue.h"
#include "TaskQueue.cpp"

template <typename T>
class ThreadPool
{
public:
    ThreadPool(int min, int max);
    ~ThreadPool();

    // 添加任务
    void addTask(Task<T> task);
    // 获取忙线程的个数
    int getBusyNumber();
    // 获取活着的线程个数
    int getAliveNumber();

private:
    // 工作的线程的任务函数
    static void* worker(void* arg);
    // 管理者线程的任务函数
    static void* manager(void* arg);
    void threadExit();

    static const int NUMBER = 2;

    pthread_mutex_t mutexPool;
    pthread_cond_t notEmpty;
    pthread_t* threadIDs;
    pthread_t managerID;
    TaskQueue<T>* taskQ;
    int minNum;
    int maxNum;
    int busyNum;
    int liveNum;
    int exitNum;
    bool shutdown = false;
};

线程池定义

#include "ThreadPool.h"
#include <iostream>
#include <string.h>
#include <string>
#include <unistd.h>

using namespace std;

template <typename T>
ThreadPool<T>::ThreadPool(int min, int max) {
    // 实例化任务队列
    do {
        taskQ = new TaskQueue<T>;
        if(taskQ == nullptr) {
            cout << "malloc threadIDs fail...\n";
            break;
        }

        threadIDs = new pthread_t[max];
        if(threadIDs == nullptr) {
            cout << "malloc threadIDs fail....\n";
            break;
        }
        memset(threadIDs, 0, sizeof(pthread_t) * max);
        minNum = min;
        maxNum = max;
        busyNum = 0;
        liveNum = min; // 和最小个数相等
        exitNum = 0;

        if(pthread_mutex_init(&mutexPool, NULL) != 0 ||
        pthread_cond_init(&notEmpty, NULL) != 0
        ) {
            cout << "mutex or condif init fail....\n";
            break;
        }

        shutdown = false;
        
        /*
        为什么要把this传给manager呢？
        因为manager是一个静态方法，静态方法它只能访问类里边的静态成员变量，
        它不能访问类的非静态成员变量。
        因此如果我们想要访问这些非静态成员变量，就必须要给这个静态方法传进去一个
        实例化对象，通过传进去的这个实例化对象来访问里边的非静态成员函数和变量
        */
        // 创建管理者线程和工作者线程
        pthread_create(&managerID, NULL, manager, this); // 第三个参数为管理者线程的任务函数
        for(int i = 0; i < min; i++) {
            pthread_create(&threadIDs[i], NULL, worker, this); // 第三个参数为工作的线程的任务函数
        }
        return;
    } while(0);

    // 释放资源
    if(threadIDs) delete []threadIDs;
    if(taskQ) delete taskQ;
}

template <typename T>
ThreadPool<T>::~ThreadPool() {

    // 关闭线程池
    shutdown = true;
    // 阻塞回收管理者线程
    pthread_join(managerID, NULL);
    // 唤醒阻塞的消费者线程
    for(int i = 0; i < liveNum; i++) { // 因为到了最后没有任务了，所以活着的线程都需要关闭
        pthread_cond_signal(&notEmpty);
    }
    // 释放堆内存
    if(taskQ) {
        delete taskQ;
    }
    if(threadIDs) {
        delete []threadIDs;
    }
    
    pthread_mutex_destroy(&mutexPool);
    pthread_cond_destroy(&notEmpty);
}

template <typename T>
void ThreadPool<T>::addTask(Task<T> task) {
    if(shutdown) {
        return;
    }
    // 添加任务
    taskQ->addTask(task);

    pthread_cond_signal(&notEmpty); // 唤醒阻塞在条件变量里的工作的线程，即生产者生产后唤醒消费者
}

template <typename T>
int ThreadPool<T>::getBusyNumber() {
    pthread_mutex_lock(&mutexPool);
    int busyNum = this->busyNum;
    pthread_mutex_unlock(&mutexPool);
    return busyNum;
}

template <typename T>
int ThreadPool<T>::getAliveNumber() {
    pthread_mutex_lock(&mutexPool);
    int liveNum = this->liveNum;
    pthread_mutex_unlock(&mutexPool);
    return liveNum;
}

template <typename T>
void* ThreadPool<T>::worker(void* arg) {
    ThreadPool* pool = static_cast<ThreadPool*>(arg); // static_cast相当于c里面的强制类型转换
    while(1) {
        pthread_mutex_lock(&pool->mutexPool); // 访问线程池之前加锁
        // 当前任务队列是否为空
        while(pool->taskQ->taskNumber() == 0 && !pool->shutdown) { // 任务队列为空并且线程池没有被关闭
            // 阻塞工作线程
            pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);

            // 判断是不是要销毁线程
            if(pool->exitNum > 0) {
                pool->exitNum--;
                if(pool->liveNum > pool->minNum) {
                    pool->liveNum--;
                    pthread_mutex_unlock(&pool->mutexPool); // 阻塞的时候已经获得锁，已经锁上，如果不解开就会死锁
                    pool->threadExit();
                }
            }
        }

        // 判断线程池是否被关闭了
        if(pool->shutdown) {
            pthread_mutex_unlock(&pool->mutexPool); // 避免死锁
            pool->threadExit();
        }

        // 从任务队列中取出一个任务
        Task<T> task = pool->taskQ->takeTask();

        pool->busyNum++;
        // 因为任务队列可以无限大，所以生产者可以不用唤醒
        pthread_mutex_unlock(&pool->mutexPool); // 用完之后解锁

        printf("thread %ld start working...\n", pthread_self());
        task.function(task.arg);
        delete task.arg; // 
        task.arg = nullptr;

        printf("thread %ld end working...\n", pthread_self());

        pthread_mutex_lock(&pool->mutexPool);
        pool->busyNum--;
        pthread_mutex_unlock(&pool->mutexPool);
    }
    return NULL;
}

template <typename T>
void* ThreadPool<T>::manager(void* arg) {
    ThreadPool* pool = static_cast<ThreadPool*>(arg);
    while(!pool->shutdown) {
        // 每隔三秒钟检测一次
        sleep(3);

        // 取出线程池中任务的数量和当前线程的数量
        pthread_mutex_lock(&pool->mutexPool);
        int queueSize = pool->taskQ->taskNumber();
        int liveNumber = pool->liveNum;
        int busyNumber = pool->busyNum; // 取出忙的线程的数量
        pthread_mutex_unlock(&pool->mutexPool);
        

        // 添加线程
        // 任务的个数 大于 存活的线程个数 && 存活的线程数 小于 最大线程数
        if(queueSize > liveNumber && liveNumber < pool->maxNum) {
            pthread_mutex_lock(&pool->mutexPool);
            int count = 0;
            for(int i = 0; i < pool->maxNum && count < NUMBER && liveNumber < pool->maxNum; i++) {
                if(pool->threadIDs[i] == 0) {
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool);
                    count++;
                    pool->liveNum++;
                }
            }
            pthread_mutex_unlock(&pool->mutexPool);
        }

        // 销毁线程
        // 忙的线程 * 2 小于 存活的线程数 && 存活的线程 大于 最小线程数
        // 之所以不用对pool->minNum加锁，是因为这个值是固定的，只需要读不需要写
        if(busyNumber * 2 < liveNumber && liveNumber > pool->minNum) {
            pthread_mutex_lock(&pool->mutexPool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutexPool);
            // 让工作的线程自杀
            for(int i = 0; i < NUMBER; i++) {
                // 唤醒已经被阻塞的工作的线程，因为工作的线程无事可做，肯定是被阻塞在条件变量中
                pthread_cond_signal(&pool->notEmpty); 
            }
        }
    }
}

template <typename T>
void ThreadPool<T>::threadExit() {
    pthread_t tid = pthread_self(); // 获得当前线程的线程ID。
    for(int i = 0; i < maxNum; i++) {
        if(threadIDs[i] == tid) { // tid对应的线程要退出了
            threadIDs[i] = 0;
            printf("threadExit() called, %ld exiting...\n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}

测试

#include "ThreadPool.h"
#include "ThreadPool.cpp"
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>

void taskFunc(void* arg) {
    int num = *(int*)arg;
    printf("thread %ld is working, number = %d\n", pthread_self(), num);
    sleep(1);
}

int main() {
    // 创建线程池
    ThreadPool<int> pool(3, 10);
    for(int i = 0; i < 100; i++) {
        int* num = new int(i + 100);
        pool.addTask(Task<int>(taskFunc, num));
    }

    sleep(20); // 等待子线程把任务处理完毕

    return 0;
}

g++ main.cpp -o main -lpthread

原文地址：https://blog.csdn.net/liupang14159/article/details/142490119

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Linux 中 man手册基础与 man.vim 和 ctags的安装使用
下一篇：汽车总线详解

socket.io-client实现实前后端时通信功能
主体功能初步实现，后期可以优化方向，设计数据库将添加过的好友存储在数据表中，添加好友要先通过才能添加。这里我使用的后端基于node.js的koa框架前端使用的是vite。
阅读更多2024-09-25
研一奖学金计划2024/9/23有感
研一开学三个星期了难受的很多，不知道什么时候能安定下来扎下根。烦心事之一而且是花费时间最多的就是上课，选课多没有学习的时间（科研小白基础没有，向导师汇报学习心得就是见光死），讲得快听不懂（念ppt的老
阅读更多2024-09-25
心觉：如何重塑高效学习的潜意识（1）两种方法的优缺点
比如，一万个小时定律，本质讲的就是专注，长期的专注带来质变，和中国古话讲的：种瓜得瓜，种豆得豆是一个道理，和物理学里讲的能量守恒是一样的。每天产生的海量知识，实际上99%都是低维的知识，很多只是换个说
阅读更多2024-09-25
LeetCode 面试经典150题 137.只出现一次的数字II
模2加法异或、模3加法
阅读更多2024-09-25
vue2 搜索高亮关键字
vue2 搜索高亮关键字。
阅读更多2024-09-25
【GUI设计】基于图像分割的GUI系统（3），matlab实现
前面博客我们对GUI设计做了一个详细的教程，并对图像处理GUI系统设计做了一个案例展示。本次案例是基于Matlab的图像分割GUI系统（3），用matlab实现。本次内容主要分为两部分，第一部分是本
阅读更多2024-09-25
python学习-11【图形用户界面】
进行绑定。事件序列使用。
阅读更多2024-09-25
VMware集群主机电源告警/IBM X3850 X6电源告警
VMware集群主机电源告警/IBM X3850 X6电源告警
阅读更多2024-09-25
Hive 的窗口函数详解
逻辑层是 Hive 中的窗口函数，它依赖分区和排序规则来生成每个分区中的行号。物理层：Hive 在执行时，通过MapReduce或Tez实现了分布式排序和行号分配，关键类如和负责处理窗口函数的具体逻辑
阅读更多2024-09-25
PHP 函数
在PHP中，函数使用function关键字定义。一个基本的函数包括函数名称、参数（可选）和函数体。除了内置函数外，用户还可以定义自己的函数来执行特定任务。PHP函数是构建复杂应用程序的基础。通过合理地
阅读更多2024-09-25

从头开始：构建一个基于C/C++的线程池

手搓线程池

线程池工作原理和实现

线程池工作原理

1. 线程池的基本组成：

2. 线程池的基本执行流程：

3. 线程池的核心参数：

4. 线程池的生命周期：

5. 线程池的执行策略：

相关知识点

线程与进程的比较

读写锁

互斥锁

基于C语言的线程池设计与实现

基于C++的线程池设计与实现

从头开始：构建一个基于C/C++的线程池

手搓线程池

线程池工作原理和实现

线程池工作原理

1. 线程池的基本组成：

2. 线程池的基本执行流程：

3. 线程池的核心参数：

4. 线程池的生命周期：

5. 线程池的执行策略：

相关知识点

线程与进程的比较

读写锁

互斥锁

基于C语言的线程池设计与实现

基于C++的线程池设计与实现

相关文章