你好赵伟

这个屌丝很懒，什么也没留下！

热门标签

内存池解释及线程池(Linux)实现

作者：你好赵伟 | 2024-03-23 02:06:00

踩

内存池

1.内存池

1.什么是内存池

内存池是一种内存分配方式。在真正使用内存之前，先申请分配一定数量的、大小相等的内存块留作备用。当有新的内存需求时，就从内存池中分出一部分内存块，若内存块不够再继续申请新的内存。使用内存池的优点有：速度远比 malloc / free 快，因为减少了系统调用的次数，特别是频繁申请/释放内存块的情况。

但是，使用内存池也有缺点：

1.如果预先分配的内存块过多，会浪费大量的空间；

2.如果预先分配的内存块过少，则会导致频繁地向操作系统申请新的内存块，从而降低程序效率。

2.传统内存使用的弊端

1.高并发时较小内存块使用导致系统调用频繁，降低了系统的执行效率

2.频繁使用时增加了系统内存的碎片，降低内存使用效率

3.没有垃圾回收机制，容易造成内存泄漏，导致内存枯竭

4.内存分配与释放的逻辑在程序中相隔较远时，降低程序的稳定性

3.解决的方法

1.使用组件

2.使用内存池

可以解决传统内存申请和分配的大部分问题

1.内存池提前预先分配大块内存，统一释放，极大的减少了malloc 和 free 等函数的调用。

2.内存池每次请求分配大小适度的内存块，避免了碎片的产生

3.在生命周期结束后统一释放内存，完全避免了内存泄露的产生

4.在生命周期结束后统一释放内存，避免重复释放指针或释放空指针等情况

4.内存池的设计

1.内存池的设计考虑

设计逻辑应该尽量简单，避免不同请求之间互相影响，尽量降低不同模块之间的耦合

内存池生存时间应该尽可能短，与请求或者连接具有相同的周期，减少碎片堆积和内存泄漏

2.线程池

1.什么是线程池

线程池是一种利用池化技术思想来实现的线程管理技术，主要是为了复用线程、便利地管理线程和任务、并将线程的创建和任务的执行解耦开来。线程池的优点有：控制运行的线程数量，处理过程中将任务放入队列，然后在线程创建后启动这些任务，如果线程数量超过了最大数量，超出数量的线程排队等候，等其他线程执行完毕，再从队列中取出任务来执行。

2.C++实现线程池的步骤如下：

1. 定义一个任务类。

2. 定义一个线程池类。

3. 在线程池类中定义一个任务队列。

4. 在线程池类中定义一个互斥锁和一个条件变量。

5. 在线程池类中定义一个函数，用于向任务队列中添加任务。

6. 在线程池类中定义一个函数，用于从任务队列中取出任务并执行。

7. 在主函数中创建一个线程池对象，并向其中添加任务。

3.实现原理：

1.任务队列，存储需要处理的任务，由工作的线程来处理这些任务

通过线程池提供的 API 函数，将一个待处理的任务添加到任务队列，或者从任务队列中删除

已处理的任务会被从任务队列中删除

线程池的使用者，也就是调用线程池函数往任务队列中添加任务的线程就是生产者线程

2.工作的线程(任务队列任务的消费者) ，N个

线程池中维护了一定数量的工作线程，他们的作用是是不停的读任务队列，从里边取出任务并处理

工作的线程相当于是任务队列的消费者角色

如果任务队列为空，工作的线程将会被阻塞(使用条件变量/信号量阻塞)

如果阻塞之后有了新的任务，由生产者将阻塞解除，工作线程开始工作

3.管理者线程(不处理任务队列中的任务)，1个

它的任务是周期性的对任务队列中的任务数量以及处于忙状态的工作线程个数进行检测

当任务过多的时候，可以适当的创建一些新的工作线程

当任务过少的时候，可以适当的销毁一些工作的线程

4.实现线程池使用到的函数及结构体

所使用的函数都包含在头文件 <pthread.h> 中

1.pthread_mutex_t结构体

结构体原型：


typedef union
{
  struct __pthread_mutex_s __data;
  char __size[__SIZEOF_PTHREAD_MUTEX_T];
  long int __align;
} pthread_mutex_t;
    
    这是一个结构体常量，用来创建互斥锁常量，但是创建出来的互斥锁还不可以使用，需要初始化

初始化互斥锁有两种方式：

1.静态方式：使用PTHREAD_MUTEX_INITIALIZER进行初始化


pthread_mutex_t mutex_lock=PTHREAD_MUTEX_INITIALIZER;
            在LinuxThreads实现中，pthread_mutex_t是一个结构，
                而PTHREAD_MUTEX_INITIALIZER则是一个结构常量的宏。
 
    静态初始化条件变量只能拥有默认的条件变量属性，不能设置其他条件变量属性

2.动态方式：使用函数pthread_mutex_init进行初始化


int pthread_mutex_init(pthread_mutex_t *mutex, 
                        const pthread_mutexattr_t *mutexattr) 
    作用：初始化互斥锁
    参数解释：
        参数mutex：是pthread_mutex_t创建出来互斥锁变量
        参数mutexattr：用于指定互斥锁属性如下，如果为NULL则使用缺省属性。
            1.PTHREAD_MUTEX_TIMED_NP，这是缺省值，也就是普通锁。当一个线程加锁以后，
                其余请求锁的线程将形成一个等待队列，并在解锁后按优先级获得锁。
                这种锁策略保证了资源分配的公平性。
            2.PTHREAD_MUTEX_RECURSIVE_NP，嵌套锁，允许同一个线程对同一个锁成功获得多次，
                并通过多次unlock解锁。如果是不同线程请求，则在加锁线程解锁时重新竞争。 
            3.PTHREAD_MUTEX_ERRORCHECK_NP，检错锁，如果同一个线程请求同一个锁，
                则返回EDEADLK，否则与PTHREAD_MUTEX_TIMED_NP类型动作相同。
                这样就保证当不允许多次加锁时不会出现最简单情况下的死锁。
            4.PTHREAD_MUTEX_ADAPTIVE_NP，适应锁，动作最简单的锁类型，仅等待解锁后重新竞争。
    返回值：函数执行成功会返回0，失败返回非0，
            函数成功执行后，互斥锁被初始化为未锁住态

2.pthread_cond_t结构体

结构体原型：


typedef union
{
  struct __pthread_cond_s __data;
  char __size[__SIZEOF_PTHREAD_COND_T];
  __extension__ long long int __align;
} pthread_cond_t;
    
    创建一个条件变量，条件变量要与互斥量一起使用，条件本身是由互斥量保护的。
        线程在改变条件状态之前必须首先锁住互斥量
        其他线程在获得互斥量之前不会察觉到这种改变，因为互斥量必须在锁定以后才能计算条件

条件变量的初始化与互斥锁的初始化一样，静态初始化需要使用PTHREAD_COND_INITIALIZER 宏

3.pthread_create函数

函数原型：


 
int  pthread_create(pthread_t *tidp, const  pthread_attr_t *attr,
    ( void *)(*start_rtn)( void *), void  *arg);
 
    作用：创建一个线程
    参数解释：
        参数tidp：为指向线程 标识符的 指针，创建成功的线程id存放到该参数
        参数attr：用来设置线程属性
        参数3：线程运行函数的起始地址
        参数arg：运行函数的参数
    返回值：若线程创建成功，则返回0。若线程创建失败，则返回出错编号

4.pthread_join函数

函数原型：


int pthread_join(pthread_t thread, void **retval)
    作用：阻塞等待线程退出，获取线程退出状态
    参数解释：
        thread：线程ID （注意：不是指针）；
        retval：存储线程结束状态
    返回值：成功返回0；失败返回错误号，使用strerror函数获取

5.pthread_cond_signal函数

函数原型：


int pthread_cond_signal (pthread_cond_t *__cond)
    解析：但是pthread_cond_signal在多处理器上可能同时唤醒多个线程，
当你只能让一个线程处理某个任务时，其它被唤醒的线程就需要继续
wait，而且规范要求pthread_cond_signal至少唤醒一个pthread_cond_wait上的线程，
其实有些实现为了简单在单处理器上也会唤醒多个线程. 另外，某些应用，如线程池，
pthread_cond_broadcast唤醒全部线程，但我们通常只需要一部分线程去做执行任务，
所以其它的线程需要继续wait.所以强烈推荐对pthread_cond_wait()使用while循环来做条件判断.
    作用：是发送一个信号给另外一个正在处于阻塞等待状态的线程,使其脱离阻塞状态,
继续执行.如果没有线程处在阻塞等待状态,pthread_cond_signal也会成功返回
    参数：pthread_cond_t创建出来的值，为条件变量
    返回值：成功返回0，失败返回错误编号
    该函数需要和函数pthread_cond_wait函数一起使用

6.pthread_cond_destroy函数

函数原型：


int pthread_cond_destroy (pthread_cond_t *__cond)
    作用：销毁条件变量
    参数：使用pthread_cond_t创建的条件变量，需要使用地址
    返回值：成功返回0，失败返回错误编号

7.pthread_mutex_destroy函数

函数原型：


int pthread_mutex_destroy(pthread_mutex_t *mutex);
    作用：销毁互斥锁
    参数：pthread_mutex_t创建的变量，需要使用地址
    返回值：成功返回0，失败返回错误编号

8.pthread_mutex_lock函数

函数原型：


int pthread_mutex_lock(pthread_mutex_t *mutex);
 
    作用：会阻塞线程使用被该函数锁住的内容，在操作线程池的公共资源时，需要获取到该锁的使用权。
如果这个锁此时正在被其它线程占用，那么 pthread_mutex_lock() 调用会进入到这个锁的排队队列中，
并会进入阻塞状态， 直到拿到锁之后才会返回
    参数：pthread_mutex_t创建的变量，互斥锁
    返回值：若成功返回0，否则返回错误编号

9.pthread_mutex_unlock函数

函数原型：


int pthread_mutex_unlock(pthread_mutex_t *mutex);
    
    作用：会释放被pthread_mutex_lock锁住的代码段内容，使之可以被其他线程调用
    参数：pthread_mutex_t创建的变量，互斥锁
    返回值：若成功返回0，否则返回错误编号

10.pthread_cond_wait函数

函数原型：


int pthread_cond_wait(pthread_cond_t * cond, pthread_mutex_t * mutex); 
 
    解析：pthread_cond_wait函数是一个线程同步函数，用于等待条件变量的信号。当线程调用该函数时，
它会自动释放锁并进入等待状态，直到另一个线程发出信号并通知该线程条件变量已经满足。此时，
该线程会重新获得锁并继续执行。该函数通常与pthread_cond_signal或pthread_cond_broadcast
函数一起使用，以实现线程间的同步
    参数解释：
        参数cond：要等待的条件变量
        参数mutex：被唤醒线程要获取到的锁
    返回值：成功返回0，失败返回非0

11.pthread_self函数

函数原型：


pthread_t pthread_self(void)
    pthread_self函数是一个POSIX线程函数，用于获取当前线程的线程ID。它返回一个pthread_t类型的值，
这是一个唯一标识线程的值。
    返回值：返回调用线程的线程ID

5.线程池的代码实现

1.main函数代码


 
#include "threadpool.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);
 
    //销毁线程池
    threadPoolDestroy(pool);
 
    return 0;
}

2.threadpool.h代码


 
#include<stdio.h>
#include<pthread.h>
#include<unistd.h>
#include<stdlib.h>
#include<string.h>
 
#ifndef _THREADPOOL_H
#define _THREADPOOL_H
 
typedef struct ThreadPool ThreadPool;
 
//创建线程池并初始化
ThreadPool* threadPoolCreate(int min, int max, int queuesize);
 
//销毁线程池
int threadPoolDestroy(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 //_THREADPOOL_H

3.threadpool.c代码


#include "threadpool.h"
#include<pthread.h>
 
const int NUMBER = 2;
 
//任务结构体
typedef struct Task
{
    void (*function)(void* arg);
    void* arg;
}Task;
 
//线程池结构体
struct ThreadPool
{
    Task* taskQ; //任务队列
    int queueCapa; //任务容量
    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 notEmpth;    //判断队列是否空了
    
    int shutdown;                //判断是否销毁线程池，销毁为1，不销毁为0
};
 
//创建线程池并初始化
ThreadPool* threadPoolCreate(int min, int max, int queuesize)
{
    //创建一个堆区内容
    ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
    do
    {
        if (pool == NULL)
        {
            printf("malloc pool error...\n");
            break;
        }
        //创建线程的最大数量
        pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * (unsigned long)max);
        if (pool->threadIDs == NULL)
        {
            printf("malloc threadIDs error\n");
            break;
        }
        memset(pool->threadIDs, 0, sizeof(pthread_t)*(unsigned long)max);
        //初始化一些数据
        pool->maxNum = max;
        pool->minNum = min;
        pool->liveNum = min; //和刚创建的线程数相等
        pool->exitNum = 0;
        pool->busyNum = 0;
 
        //初始化锁和条件变量
        if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 ||
            pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
            pthread_cond_init(&pool->notEmpth, NULL) != 0 ||
            pthread_cond_init(&pool->notFull, NULL) != 0)
        {
            printf("mutex or cond init error..\n");
            break;
        }
 
        //初始化任务队列
        pool->taskQ = (Task*)malloc(sizeof(Task) * queuesize);
        pool->queueCapa = 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->managerID) free(pool->managerID);
    if (pool->taskQ) free(pool->taskQ);
    if (pool) free(pool);
 
    return NULL;
}
 
int threadPoolDestroy(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->notEmpth);
    }
    //释放堆内存
    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->notEmpth);
    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->queueCapa && !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->queueCapa;
    pool->queueSize++;//任务队列加1
    
    //需要唤醒阻塞在条件变量上工作线程区执行任务
    pthread_cond_signal(&pool->notEmpth);
 
    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)
{
 
 
    struct ThreadPool* pool = (struct ThreadPool*)arg;
    while (1)
    {
        //使用线程池中的资源需要先加锁
        pthread_mutex_lock(&pool->mutexPool);//加锁
        //当任务队列为空，循环线程队列
        while (pool->queueSize == 0 && !pool->shutdown)
        {
            //阻塞工作线程
            pthread_cond_wait(&pool->notEmpth, &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->queueCapa;
        pool->queueSize--;//任务个数减1
 
        //取出一个任务后，需要唤醒生产者线程，进行添加任务
        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++;//忙线程数量加1
        pthread_mutex_unlock(&pool->mutexBusy);
 
        //开始执行任务
        task.function(task.arg);//直接使用函数指针调用任务函数
        //(*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--;//忙线程数量减1
        pthread_mutex_unlock(&pool->mutexBusy);
    }
 
 
    return NULL;
}
 
//管理者线程
void* manager(void* arg)
{
    struct ThreadPool* pool = (struct ThreadPool*)arg;
    while (!pool->shutdown)//当线程池没有被关闭
    {
        sleep(3);//每隔3秒检测一次
        //取出线程池中任务的数量和当前线程池的数量
        pthread_mutex_lock(&pool->mutexPool);//上锁
        int queueSize = pool->queueSize; //取出线程池中线程的数量
        int liveNum = pool->liveNum;    //取出线程存活的数量
        pthread_mutex_unlock(&pool->mutexPool);//解锁
 
        //取出忙线程的数量
        pthread_mutex_lock(&pool->mutexBusy);
        int busyNum = pool->busyNum;
        pthread_mutex_unlock(&pool->mutexBusy);
 
        //添加线程
        //添加原则：任务的个数>存活的线程个数  &&  存活的线程数 < 最大线程数  并且每次只加连个线程
        if (queueSize > liveNum && liveNum < pool->maxNum)
        {
            pthread_mutex_lock(&pool->mutexPool);//加锁
            int counter = 0;
            for (int i = 0; i < pool->maxNum && counter < NUMBER && pool->liveNum < pool->maxNum; i++)
            {
                if (pool->threadIDs[i] == 0)//只要线程数组中为空，就创建线程
                {
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool);//创建线程
                    pool->liveNum++;//存活数加1
                    counter++;
                }
            }
            pthread_mutex_unlock(&pool->mutexPool);
        }
 
        //销毁线程
        //销毁原则：忙的线程数*2 < 存活的线程数 && 存活的线程 > 最小线程数，每次销毁还是销毁两个
        if (busyNum * 2 < liveNum && pool->minNum < liveNum)
        {
            pthread_mutex_lock(&pool->mutexPool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutexPool);
            //让工作的线程自杀
            for (int i = 0; i < NUMBER; i++)
            {
                pthread_cond_signal(&pool->notEmpth);//唤醒线程
            }
        }
 
    }
    return NULL;
}
 
void threadExit(ThreadPool* pool)
{
    pthread_t tid = pthread_self();//获取线程id
    for (int i = 0; i < pool->maxNum; i++)
    {
        if (pool->threadIDs[i] == tid)
        {
            pool->threadIDs[i] = 0;
            printf("threadExit() called, %ld exiting...\n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/292419