介绍
C Linux实现线程池技术
作者第一次编写的线程池，推荐使用的时候修改thread_manager函数中部分逻辑
支持库
代码语言：c复制#include <stdlib.h>
#include <pthread.h>
#include <string.h>
#include <time.h>
#include <unistd.h>代码
ThreadPool.h
代码语言：c复制typedef struct MissionAttr
{
void (*mission)(void *);
void *param;
time_t createTime;
char state;
} MissionAttr;
typedef struct MissionNode
{
MissionAttr *mission;
struct MissionNode *next;
struct MissionNode *back;
} MissionNode;
typedef struct MissionList
{
MissionNode *front;
MissionNode *final;
} MissionList;
typedef struct ThreadAttr
{
pthread_t id;
time_t createTime;
char state;
} ThreadAttr;
typedef struct ThreadNode
{
ThreadAttr *threadAttr;
struct ThreadNode *next;
struct ThreadNode *back;
} ThreadNode;
typedef struct ThreadList
{
ThreadNode *front;
ThreadNode *final;
} ThreadList;
typedef struct ThreadPool
{
int controlThreadId;
ThreadList *threadList;
MissionList *missionWaiting;
MissionList *missionList;
pthread_t managerID;
int waitingMissionNumber;
int hanldingMissionNumber;
int busyNumber;
int freeNumber;
int needCloseNumber;
int maxNumber;
int minNumber;
int existNumber;
char close;
pthread_mutex_t managerMutex;
pthread_mutex_t missionMutex;
pthread_mutex_t threadMutex;
pthread_mutex_t workerMutex;
pthread_cond_t workerCond;
} ThreadPool;
typedef struct ThreadArgs
{
ThreadPool *threadPool;
ThreadNode *threadNode;
} ThreadArgs;
// 主要函数
// 创建线程池
ThreadPool *create_thread_pool(int minNumber, int maxNumber);
// 提交任务
void thread_pool_submit(ThreadPool *threadPool, void *func, void *args);
// 启动线程池
int thread_pool_run(ThreadPool *threadPool);
// 关闭并释放线程池
void thread_shutdown_and_free(ThreadPool *threadPool);
// 其他函数
// 管理者线程
void *thread_manager(void *args);
// 工作者线程
void *thread_worker(void *args);
// 创建工作者线程
int creater_thread_worker(ThreadPool *threadPool, int number);
// 获取等待中的任务
MissionNode *get_mission(ThreadPool *threadPool);
// 释放完成的任务
void free_mission(ThreadPool *threadPool, MissionNode *missionNode);
// 基础函数
// 申请内存修复版
void *fixMalloc(size_t size);ThreadPool.c
代码语言：c复制#include "./ThreadPool.h"
void *fixMalloc(size_t size)
{
void *tmp = NULL;
while (1)
{
tmp = malloc(size);
if (tmp)
{
break;
}
sleep(1);
}
return tmp;
}
ThreadPool *create_thread_pool(int minNumber, int maxNumber)
{
ThreadPool *threadPool = (ThreadPool *)fixMalloc(sizeof(ThreadPool));
memset(threadPool, 0, sizeof(ThreadPool));
threadPool->minNumber = minNumber;
threadPool->maxNumber = maxNumber;
pthread_mutex_init(&threadPool->missionMutex, NULL);
pthread_mutex_init(&threadPool->threadMutex, NULL);
pthread_mutex_init(&threadPool->workerMutex, NULL);
pthread_cond_init(&threadPool->workerCond, NULL);
pthread_mutex_init(&threadPool->managerMutex, NULL);
threadPool->threadList = (ThreadList *)fixMalloc(sizeof(ThreadList));
threadPool->missionList = (MissionList *)fixMalloc(sizeof(MissionList));
threadPool->missionWaiting = (MissionList *)fixMalloc(sizeof(MissionList));
memset(threadPool->threadList, 0, sizeof(ThreadList));
memset(threadPool->missionList, 0, sizeof(MissionList));
memset(threadPool->missionWaiting, 0, sizeof(ThreadList));
threadPool->controlThreadId = 0;
threadPool->waitingMissionNumber = 0;
threadPool->hanldingMissionNumber = 0;
threadPool->busyNumber = 0;
threadPool->freeNumber = 0;
threadPool->needCloseNumber = 0;
threadPool->existNumber = 0;
return threadPool;
}
void thread_pool_submit(ThreadPool *threadPool, void *func, void *args)
{
if (!threadPool->close && func != NULL)
{
MissionList *missionWaiting = threadPool->missionWaiting;
MissionNode *newMissionNode = (MissionNode *)fixMalloc(sizeof(MissionNode));
newMissionNode->mission = (MissionAttr *)fixMalloc(sizeof(MissionAttr));
newMissionNode->mission->createTime = time(NULL);
newMissionNode->mission->mission = func;
newMissionNode->mission->param = args;
newMissionNode->mission->state = 0;
newMissionNode->next = NULL;
if (missionWaiting->final == NULL)
{
missionWaiting->front = newMissionNode;
}
else
{
newMissionNode->back = missionWaiting->final;
missionWaiting->final->next = newMissionNode;
}
missionWaiting->final = newMissionNode;
threadPool->waitingMissionNumber = threadPool->waitingMissionNumber 1;
}
}
MissionNode *get_mission(ThreadPool *threadPool)
{
MissionList *missionWaiting = threadPool->missionWaiting;
if (missionWaiting->front == NULL)
{
return NULL;
}
MissionList *missionList = threadPool->missionList;
MissionNode *missionNode = missionWaiting->front;
if (missionWaiting->front == missionWaiting->final)
{
missionWaiting->final = NULL;
}
missionWaiting->front = missionNode->next;
if (missionNode->next)
{
missionNode->next->back = NULL;
}
missionNode->next = NULL;
if (missionList->front == NULL)
{
missionList->front = missionNode;
}
else
{
missionNode->back = missionList->final;
missionList->final->next = missionNode;
}
missionList->final = missionNode;
return missionNode;
}
void free_mission(ThreadPool *threadPool, MissionNode *missionNode)
{
MissionList *missionList = threadPool->missionList;
if (missionNode->back == NULL)
{
missionList->front = missionNode->next;
}
if (missionNode->back)
{
missionNode->back->next = missionNode->next;
}
if (missionNode->next)
{
missionNode->next->back = missionNode->back;
}
else
{
missionList->final = missionNode->back;
}
if (missionNode->mission)
free(missionNode->mission);
if (missionNode)
free(missionNode);
}
void *thread_worker(void *args)
{
ThreadArgs *threadArgs = (ThreadArgs *)args;
ThreadPool *threadPool = threadArgs->threadPool;
ThreadNode *selfNode = threadArgs->threadNode;
threadPool->existNumber = threadPool->existNumber 1;
threadPool->freeNumber = threadPool->freeNumber 1;
selfNode->threadAttr->state = 1;
pthread_mutex_lock(&threadPool->workerMutex);
while (threadPool->needCloseNumber <= 0 && !threadPool->close)
{
// 等待任务
pthread_cond_wait(&threadPool->workerCond, &threadPool->workerMutex);
pthread_mutex_unlock(&threadPool->workerMutex);
if (threadPool->waitingMissionNumber > 0)
{
pthread_mutex_lock(&threadPool->missionMutex);
threadPool->freeNumber = threadPool->freeNumber - 1;
selfNode->threadAttr->state = 2;
if (threadPool->close)
break;
MissionNode *missionNode = get_mission(threadPool);
MissionAttr *missionAttr = NULL;
if (missionNode != NULL)
{
missionAttr = missionNode->mission;
}
selfNode->threadAttr->state = 3;
if (missionAttr != NULL)
{
missionAttr->state = 1;
}
threadPool->busyNumber = threadPool->busyNumber 1;
threadPool->waitingMissionNumber = threadPool->waitingMissionNumber - 1;
threadPool->hanldingMissionNumber = threadPool->hanldingMissionNumber 1;
if (threadPool->close)
break;
pthread_mutex_unlock(&threadPool->missionMutex);
if (missionAttr != NULL)
if (missionAttr->mission)
{
missionAttr->mission(missionAttr->param);
}
pthread_mutex_lock(&threadPool->missionMutex);
if (threadPool->close)
break;
if (missionAttr != NULL)
if (missionAttr->mission)
{
free_mission(threadPool, missionNode);
}
threadPool->hanldingMissionNumber = threadPool->hanldingMissionNumber - 1;
threadPool->busyNumber = threadPool->busyNumber - 1;
threadPool->freeNumber = threadPool->freeNumber 1;
selfNode->threadAttr->state = 1;
pthread_mutex_unlock(&threadPool->missionMutex);
}
}
if (threadPool->close)
{
pthread_mutex_unlock(&threadPool->workerMutex);
pthread_exit(NULL);
}
// 线程自杀
pthread_mutex_lock(&threadPool->threadMutex);
ThreadList *threadList = threadPool->threadList;
selfNode->threadAttr->state = -1;
if (threadList->final)
if (threadList->final->threadAttr)
if (threadList->final->threadAttr->id == selfNode->threadAttr->id)
{
threadList->final = selfNode->back;
}
if (threadList->front)
if (threadList->front->threadAttr)
if (threadList->front->threadAttr->id == selfNode->threadAttr->id)
{
threadList->front = selfNode->next;
}
if (selfNode->back)
if (selfNode->back->next)
{
selfNode->back->next = selfNode->next;
}
if (selfNode->next)
if (selfNode->next->back)
{
selfNode->next->back = selfNode->back;
}
free(selfNode->threadAttr);
free(selfNode);
threadPool->existNumber = threadPool->existNumber - 1;
threadPool->freeNumber = threadPool->freeNumber - 1;
threadPool->needCloseNumber = threadPool->needCloseNumber - 1;
pthread_mutex_unlock(&threadPool->threadMutex);
pthread_mutex_unlock(&threadPool->workerMutex);
pthread_exit(NULL);
}
int creater_thread_worker(ThreadPool *threadPool, int number)
{
if ((number threadPool->existNumber) > threadPool->maxNumber)
{
number = threadPool->maxNumber - threadPool->existNumber;
}
pthread_mutex_lock(&threadPool->threadMutex);
int i, ref;
for (i = 0; i < number; i )
{
pthread_t threadID = 0;
ThreadNode *newThreadNode = (ThreadNode *)fixMalloc(sizeof(ThreadNode));
ThreadArgs *threadArgs = (ThreadArgs *)fixMalloc(sizeof(ThreadArgs));
// 设置Woker线程参数
threadArgs->threadNode = newThreadNode;
threadArgs->threadPool = threadPool;
// 初始化线程节点
newThreadNode->threadAttr = (ThreadAttr *)fixMalloc(sizeof(ThreadAttr));
newThreadNode->back = NULL;
newThreadNode->next = NULL;
// 设置线程属性
newThreadNode->threadAttr->createTime = time(NULL);
newThreadNode->threadAttr->id = threadID;
newThreadNode->threadAttr->state = 0;
// 插入线程链表
if (threadPool->existNumber == 0)
{
threadPool->threadList->front = newThreadNode;
}
else
{
ThreadNode *threadNode = threadPool->threadList->final;
if (threadNode)
{
threadNode->next = newThreadNode;
}
newThreadNode->back = threadNode;
}
threadPool->threadList->final = newThreadNode;
ref = pthread_create(&threadID, NULL, (void *)thread_worker, (void *)threadArgs);
if (ref != 0)
{
return -2;
}
}
pthread_mutex_unlock(&threadPool->threadMutex);
return 0;
}
void *thread_manager(void *args)
{
// 管理者
ThreadPool *threadPool = (ThreadPool *)args;
int minNumber = threadPool->minNumber;
int waitingMissionNumber = 0;
int existNumber = 0;
int needCloseNumber = 0;
int busyNumber = 0;
int runNumber = 0;
int i;
pthread_t threadID;
creater_thread_worker(threadPool, minNumber);
while (!threadPool->close)
{
pthread_mutex_lock(&threadPool->managerMutex);
waitingMissionNumber = threadPool->waitingMissionNumber;
existNumber = threadPool->existNumber;
needCloseNumber = threadPool->needCloseNumber;
busyNumber = threadPool->busyNumber;
pthread_mutex_unlock(&threadPool->managerMutex);
// 这里的逻辑写的很随意，推荐使用时重写此逻辑
if (waitingMissionNumber > 0)
{
if (waitingMissionNumber > existNumber)
{
creater_thread_worker(threadPool, 2);
}
runNumber = existNumber < waitingMissionNumber ? existNumber : waitingMissionNumber;
for (i = 0; i < runNumber; i )
{
pthread_cond_signal(&threadPool->workerCond);
}
}
else
{
threadPool->needCloseNumber = existNumber - threadPool->minNumber;
pthread_cond_broadcast(&threadPool->workerCond);
}
usleep(1000 * 100);
}
// 唤醒所有线程
pthread_cond_broadcast(&threadPool->workerCond);
// 释放Worker线程
ThreadList *threadList = threadPool->threadList;
ThreadNode *threadNode = threadList->front;
ThreadNode *tmpThreadNode = NULL;
while (threadNode != NULL)
{
threadID = threadNode->threadAttr->id;
if (threadID)
{
pthread_join(threadID, NULL);
}
free(threadNode->threadAttr);
tmpThreadNode = threadNode->next;
free(threadNode);
threadNode = tmpThreadNode;
}
free(threadList);
MissionList *missionWaiting = threadPool->missionWaiting;
MissionNode *missionWaitingNode = missionWaiting->front;
MissionNode *tmpMissionWaitingNode = NULL;
// 释放等待列队
while (missionWaitingNode != NULL)
{
free(missionWaitingNode->mission);
tmpMissionWaitingNode = missionWaitingNode->next;
free(missionWaitingNode);
missionWaitingNode = tmpMissionWaitingNode;
}
free(missionWaiting);
// 释放任务列队
MissionList *missionList = threadPool->missionList;
MissionNode *missionNode = missionList->front;
MissionNode *tmpMissionNode = NULL;
while (missionNode != NULL)
{
free(missionNode->mission);
tmpMissionNode = missionNode->next;
free(missionNode);
missionNode = tmpMissionNode;
}
free(missionList);
pthread_mutex_destroy(&threadPool->missionMutex);
pthread_mutex_destroy(&threadPool->workerMutex);
pthread_mutex_destroy(&threadPool->threadMutex);
pthread_mutex_destroy(&threadPool->managerMutex);
pthread_cond_destroy(&threadPool->workerCond);
free(threadPool);
pthread_exit(NULL);
}
int thread_pool_run(ThreadPool *threadPool)
{
int ref = pthread_create(&threadPool->managerID, NULL, thread_manager, (void *)threadPool);
return ref;
}
void thread_shutdown_and_free(ThreadPool *threadPool)
{
threadPool->close = 1;
pthread_join(threadPool->managerID, NULL);
}示例
代码语言：c复制#include <stdio.h>
#include "./ThreadPool.c"
struct testData
{
int a;
};
ThreadPool *pool = NULL;
int times = 0;
void test(void *a)
{
// 模拟工作
// 处理参数
int *b = (int *)a;
times ;
sleep(*b);
}
int main()
{
// 初始化线程池
pool = create_thread_pool(1, 50);
// 启动线程池
thread_pool_run(pool);
int i, a = 1;
// 提交任务
for (i = 0; i < 110; i )
{
// 指针传参（无参数填NULL）
thread_pool_submit(pool, &test, &a);
}
i = 0;
while (i < 52)
{
printf("当前任务数量:%dn", pool->waitingMissionNumber);
printf("当前线程数量:%dn", pool->existNumber);
printf("当前执行线程:%dn", pool->busyNumber);
printf("当前空闲线程:%dn", pool->freeNumber);
printf("当前执行任务:%dn", pool->hanldingMissionNumber);
printf("当前自杀线程:%dn", pool->needCloseNumber);
printf("函数执行次数:%dn", times);
printf("================================n");
sleep(1);
int rands = (rand() / 1000000000.0) * 20 10;
printf("当前随机数:%dn", rands);
// 模拟不定时提交
if (rands > 40)
for (i = 0; i < rands; i )
{
thread_pool_submit(pool, &test, &a);
}
}
// 关机并释放
thread_shutdown_and_free(pool);
return 0;
}编译命令示例
代码语言：shell复制gcc ./main.c -o ./test.out -lpthread
