Linux进程池与线程池深度解析：设计原理+实战实现（网盘项目架构）

Linux进程池与线程池深度解析：设计原理+实战实现（网盘项目架构）

前言

在高并发场景（如网盘文件传输、Web服务器）中，频繁创建/销毁进程/线程会导致大量资源开销，严重影响程序性能。池化技术（进程池/线程池）是解决该问题的核心方案——程序启动时预先创建固定数量的进程/线程，闲置时阻塞等待任务，任务完成后重回空闲状态，避免动态创建/销毁的开销。

本文以网盘文件传输项目为背景，从设计思路、整体架构、核心实现（父子进程通信、文件描述符传递）到完整实战代码，全面讲解Linux进程池的实现逻辑，同时延伸线程池的核心差异与应用场景，帮你掌握高并发程序的架构设计与开发技巧。

前置知识：Linux进程/线程管理、Socket编程、进程间通信（IPC）、C语言基础
核心技术：fork/socketpair/sendmsg/recvmsg、epoll IO多路复用、状态管理、任务分发
适用场景：高并发服务器、文件传输系统、API网关等需要频繁处理任务的场景

一、进程池与线程池的设计背景

1.1 传统多进程模型的痛点

在网盘等需要处理大量客户端请求的场景中，传统“一请求一进程”模型存在明显缺陷：

• 资源开销大：进程创建/销毁需要内核分配/释放地址空间、PCB等资源，耗时占比高；
• 响应延迟高：大量时间消耗在进程切换和资源调度上，而非核心业务（文件传输）；
• 可扩展性差：进程数量过多会导致CPU上下文切换频繁，系统负载飙升。

1.2 池化技术的核心思想

“池化”本质是资源预分配与复用：

1.3 进程池vs线程池：选型建议

二、进程池的整体架构（网盘项目实战）

以网盘文件下载功能为例，进程池架构分为主进程（Master） 和工作进程（Worker） 两部分，配合请求队列和IO多路复用实现高效任务分发：

2.1 架构流程图

客户端(Client) → 请求队列 → 主进程(Master) → 工作进程(Worker) → 磁盘存储(Disk)
↑↓
任务分发/状态管理
（epoll监听）

2.2 核心组件职责

1. 主进程（Master）

• 初始化：创建工作进程池、绑定网络端口、初始化epoll实例；
• 连接管理：通过epoll监听客户端TCP连接和工作进程的通信套接字；
• 任务分发：接收客户端连接后，选择空闲（FREE）工作进程，传递任务（已连接套接字+操作指令）；
• 状态监控：维护工作进程的状态（FREE/BUSY），处理工作进程的状态反馈。

2. 工作进程（Worker）

• 初始化：创建后进入空闲状态，通过本地套接字等待主进程任务通知；
• 任务处理：接收主进程传递的已连接套接字和指令（如下载文件路径），与客户端交互完成文件传输；
• 状态更新：任务完成后，通过本地套接字通知主进程，重回FREE状态。

3. 请求队列（可选）

• 当客户端请求数量超过工作进程数时，暂存请求，避免请求丢失；
• 主进程按FIFO（先进先出）原则，依次向空闲工作进程分发任务。

三、进程池的核心实现（完整代码实战）

以下基于网盘文件下载场景，实现一个支持高并发的进程池，核心步骤：父子进程创建、主进程epoll监听、父子进程通信（文件描述符传递）、任务分发与处理。

3.1 头文件与全局定义

// head.h
#ifndef __PROCESS_POOL_H__
#define __PROCESS_POOL_H__

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/wait.h>
#include <sys/types.h>
#include <errno.h>
#include <sys/uio.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/epoll.h>

// 工作进程状态：空闲/忙碌
typedef enum {
FREE = 0,
BUSY
} WorkerStatus;

// 工作进程信息结构体：存储PID、状态和通信套接字
typedef struct {
pid_t pid; // 工作进程PID
WorkerStatus status; // 工作进程状态
int pipeFd; // 与主进程通信的本地套接字文件描述符
} ProcessData;

// 错误检查宏
#define ERROR_CHECK(ret, val, msg) { \\
if (ret == val) { \\
perror(msg); \\
exit(1); \\
} \\
}

// 命令行参数检查宏
#define ARGS_CHECK(argc, n) { \\
if (argc != n) { \\
fprintf(stderr, "用法：%s 服务器IP 端口号 进程池大小\\n", argv[0]); \\
exit(1); \\
} \\
}

// epoll相关宏定义
#define MAX_EVENTS 1024 // 最大监听事件数
#define EPOLL_TIMEOUT –1 // epoll阻塞等待（无超时）

// 函数声明
int tcpInit(char *ip, char *port, int *pSockFd); // TCP初始化（socket+bind+listen）
int makeChild(ProcessData *workerList, int workerNum); // 创建工作进程池
void workerHandle(int pipeFd); // 工作进程任务处理
int epollCtor(int *pEpfd); // 初始化epoll实例
int epollAdd(int fd, int epfd); // 添加文件描述符到epoll
int epollDel(int fd, int epfd); // 从epoll删除文件描述符
void masterEpollLoop(int sockFd, int epfd, ProcessData *workerList, int workerNum); // 主进程epoll循环
ssize_t sendFd(int sockFd, int fd); // 传递文件描述符（主→工）
ssize_t recvFd(int sockFd, int *fd); // 接收文件描述符（工→主）

#endif

3.2 核心工具函数实现

3.2.1 TCP初始化（主进程监听网络连接）

// tcpInit.c
#include "head.h"

int tcpInit(char *ip, char *port, int *pSockFd) {
// 1. 创建TCP套接字
int sockFd = socket(AF_INET, SOCK_STREAM, 0);
ERROR_CHECK(sockFd, –1, "socket");

// 2. 设置地址重用（解决TIME_WAIT端口占用）
int reuse = 1;
int ret = setsockopt(sockFd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
ERROR_CHECK(ret, –1, "setsockopt");

// 3. 配置服务器地址
struct sockaddr_in addr;
bzero(&addr, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = inet_addr(ip);
addr.sin_port = htons(atoi(port));

// 4. 绑定IP和端口
ret = bind(sockFd, (struct sockaddr *)&addr, sizeof(addr));
ERROR_CHECK(ret, –1, "bind");

// 5. 监听连接（全连接队列大小10）
ret = listen(sockFd, 10);
ERROR_CHECK(ret, –1, "listen");

*pSockFd = sockFd;
return 0;
}

3.2.2 epoll初始化与事件管理

// epoll_func.c
#include "head.h"

// 初始化epoll实例
int epollCtor(int *pEpfd) {
int epfd = epoll_create(1); // size参数已废弃，填>0即可
ERROR_CHECK(epfd, –1, "epoll_create");
*pEpfd = epfd;
return 0;
}

// 添加文件描述符到epoll（水平触发LT）
int epollAdd(int fd, int epfd) {
struct epoll_event ev;
ev.events = EPOLLIN; // 监听读事件
ev.data.fd = fd;
int ret = epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);
ERROR_CHECK(ret, –1, "epoll_ctl add");
return 0;
}

// 从epoll删除文件描述符
int epollDel(int fd, int epfd) {
int ret = epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
ERROR_CHECK(ret, –1, "epoll_ctl del");
return 0;
}

3.2.3 父子进程通信：传递文件描述符

进程池的核心难点是主进程将客户端连接的netFd传递给工作进程（进程地址空间隔离，无法直接共享）。解决方案：使用socketpair创建本地套接字（全双工管道），配合sendmsg/recvmsg传递文件描述符。

关键原理

• socketpair：创建一对相互连接的本地套接字，用于父子进程双向通信；
• sendmsg/recvmsg：支持传递“辅助数据”（控制信息），文件描述符通过辅助数据传递；
• struct cmsghdr：存储辅助数据（文件描述符），需通过CMSG_*宏操作，避免直接修改。

实现代码

// fd_transfer.c
#include "head.h"

// 发送文件描述符：sockFd=通信套接字，fd=待传递的文件描述符
ssize_t sendFd(int sockFd, int fd) {
struct msghdr msg;
bzero(&msg, sizeof(msg));

// 1. 普通数据（可选，此处传递操作指令）
char buf[10] = "download";
struct iovec iov;
iov.iov_base = buf;
iov.iov_len = strlen(buf);
msg.msg_iov = &iov;
msg.msg_iovlen = 1;

// 2. 辅助数据：存储文件描述符
struct cmsghdr *cmsg;
char cmsgBuf[CMSG_SPACE(sizeof(int))]; // 辅助数据缓冲区
msg.msg_control = cmsgBuf;
msg.msg_controllen = CMSG_SPACE(sizeof(int));

cmsg = CMSG_FIRSTHDR(&msg); // 获取第一个辅助数据
cmsg->cmsg_level = SOL_SOCKET; // 协议级别（套接字层）
cmsg->cmsg_type = SCM_RIGHTS; // 类型：传递文件描述符
cmsg->cmsg_len = CMSG_LEN(sizeof(int)); // 数据长度
*(int *)CMSG_DATA(cmsg) = fd; // 写入文件描述符

// 3. 发送消息（包含普通数据和文件描述符）
ssize_t ret = sendmsg(sockFd, &msg, 0);
ERROR_CHECK(ret, –1, "sendmsg");
return ret;
}

// 接收文件描述符：sockFd=通信套接字，fd=输出参数（存储接收的文件描述符）
ssize_t recvFd(int sockFd, int *fd) {
struct msghdr msg;
bzero(&msg, sizeof(msg));

// 1. 接收普通数据
char buf[10];
struct iovec iov;
iov.iov_base = buf;
iov.iov_len = sizeof(buf);
msg.msg_iov = &iov;
msg.msg_iovlen = 1;

// 2. 接收辅助数据（文件描述符）
char cmsgBuf[CMSG_SPACE(sizeof(int))];
msg.msg_control = cmsgBuf;
msg.msg_controllen = CMSG_SPACE(sizeof(int));

// 3. 接收消息
ssize_t ret = recvmsg(sockFd, &msg, 0);
ERROR_CHECK(ret, –1, "recvmsg");

// 4. 解析文件描述符
struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
*fd = *(int *)CMSG_DATA(cmsg);

printf("工作进程接收：操作指令=%s，文件描述符=%d\\n", buf, *fd);
return ret;
}

3.3 进程池创建与任务处理

3.3.1 创建工作进程池（主进程创建子进程）

// process_pool.c
#include "head.h"

int makeChild(ProcessData *workerList, int workerNum) {
int sv[2]; // socketpair创建的本地套接字（sv[0]主进程用，sv[1]子进程用）
pid_t pid;

for (int i = 0; i < workerNum; i++) {
// 1. 创建本地套接字（父子进程通信）
int ret = socketpair(AF_LOCAL, SOCK_STREAM, 0, sv);
ERROR_CHECK(ret, –1, "socketpair");

// 2. fork创建子进程
pid = fork();
if (pid == 0) {
// 子进程：关闭主进程端套接字，进入任务处理循环
close(sv[0]);
workerHandle(sv[1]);
exit(0); // 子进程无需返回主进程
} else if (pid > 0) {
// 主进程：关闭子进程端套接字，记录工作进程信息
close(sv[1]);
workerList[i].pid = pid;
workerList[i].status = FREE;
workerList[i].pipeFd = sv[0];
} else {
ERROR_CHECK(pid, –1, "fork");
}
}
return 0;
}

3.3.2 工作进程任务处理（文件下载核心逻辑）

工作进程接收主进程传递的netFd（客户端连接），读取客户端请求的文件路径，将文件内容发送给客户端，完成后通知主进程重置状态：

// worker_handle.c
#include "head.h"

void workerHandle(int pipeFd) {
int netFd; // 客户端连接的文件描述符
char filePath[128]; // 客户端请求的文件路径
int fileFd; // 待下载文件的文件描述符
char buf[4096]; // 数据传输缓冲区
ssize_t ret;

while (1) {
// 1. 接收主进程传递的文件描述符和指令（阻塞等待）
recvFd(pipeFd, &netFd);

// 2. 读取客户端请求的文件路径
bzero(filePath, sizeof(filePath));
ret = recv(netFd, filePath, sizeof(filePath), 0);
ERROR_CHECK(ret, –1, "recv filePath");
if (ret == 0) {
// 客户端主动断开连接
close(netFd);
send(pipeFd, "FREE", strlen("FREE"), 0); // 通知主进程重回空闲
continue;
}
printf("工作进程PID=%d，处理文件下载：%s\\n", getpid(), filePath);

// 3. 打开文件（只读模式）
fileFd = open(filePath, O_RDONLY);
if (fileFd == –1) {
// 文件不存在，向客户端返回错误信息
send(netFd, "文件不存在", strlen("文件不存在"), 0);
close(netFd);
send(pipeFd, "FREE", strlen("FREE"), 0);
continue;
}

// 4. 向客户端发送文件内容（循环读取并发送）
while ((ret = read(fileFd, buf, sizeof(buf))) > 0) {
send(netFd, buf, ret, 0);
bzero(buf, sizeof(buf));
}

// 5. 关闭文件和客户端连接
close(fileFd);
close(netFd);

// 6. 通知主进程：任务完成，状态重置为FREE
send(pipeFd, "FREE", strlen("FREE"), 0);
printf("工作进程PID=%d，文件下载完成，重回空闲状态\\n", getpid());
}
}

3.3.3 主进程epoll循环（高并发核心）

主进程通过epoll同时监听“客户端连接”和“工作进程状态反馈”，实现高效任务分发：

// master_epoll_loop.c
#include "head.h"

// 选择空闲工作进程（轮询策略）
static int selectFreeWorker(ProcessData *workerList, int workerNum) {
for (int i = 0; i < workerNum; i++) {
if (workerList[i].status == FREE) {
return i;
}
}
return –1; // 无空闲工作进程
}

void masterEpollLoop(int sockFd, int epfd, ProcessData *workerList, int workerNum) {
struct epoll_event readyEvents[MAX_EVENTS];
struct sockaddr_in clientAddr;
socklen_t clientLen = sizeof(clientAddr);
int netFd; // 客户端连接的文件描述符
char statusBuf[10]; // 接收工作进程的状态通知

printf("主进程启动epoll循环，监听客户端连接和工作进程状态…\\n");

while (1) {
// 1. 等待epoll事件就绪（阻塞）
int readyNum = epoll_wait(epfd, readyEvents, MAX_EVENTS, EPOLL_TIMEOUT);
ERROR_CHECK(readyNum, –1, "epoll_wait");

// 2. 遍历就绪事件
for (int i = 0; i < readyNum; i++) {
int fd = readyEvents[i].data.fd;

// 3. 客户端连接事件（监听套接字就绪）
if (fd == sockFd) {
netFd = accept(sockFd, (struct sockaddr *)&clientAddr, &clientLen);
ERROR_CHECK(netFd, –1, "accept");
printf("客户端连接：IP=%s，端口=%d，netFd=%d\\n",
inet_ntoa(clientAddr.sin_addr), ntohs(clientAddr.sin_port), netFd);

// 选择空闲工作进程
int freeIdx = selectFreeWorker(workerList, workerNum);
if (freeIdx == –1) {
// 无空闲工作进程，向客户端返回提示
send(netFd, "服务器繁忙，请稍后再试", strlen("服务器繁忙，请稍后再试"), 0);
close(netFd);
printf("无空闲工作进程，拒绝客户端连接：%s:%d\\n",
inet_ntoa(clientAddr.sin_addr), ntohs(clientAddr.sin_port));
continue;
}

// 标记工作进程为BUSY，传递文件描述符
workerList[freeIdx].status = BUSY;
sendFd(workerList[freeIdx].pipeFd, netFd);
close(netFd); // 主进程无需持有netFd，由工作进程处理
}
// 4. 工作进程状态反馈事件（本地套接字就绪）
else {
bzero(statusBuf, sizeof(statusBuf));
ssize_t ret = recv(fd, statusBuf, sizeof(statusBuf), 0);
if (ret == 0) {
// 工作进程异常退出，重新创建（简化处理）
printf("工作进程异常退出，pipeFd=%d\\n", fd);
epollDel(fd, epfd);
close(fd);
continue;
}

// 工作进程完成任务，重置状态为FREE
if (strcmp(statusBuf, "FREE") == 0) {
for (int j = 0; j < workerNum; j++) {
if (workerList[j].pipeFd == fd) {
workerList[j].status = FREE;
printf("工作进程PID=%d 状态重置为FREE\\n", workerList[j].pid);
break;
}
}
}
}
}
}
}

3.4 主函数（程序入口）

// main.c
#include "head.h"

int main(int argc, char *argv[]) {
// 1. 检查命令行参数（./process_pool 服务器IP 端口号 进程池大小）
ARGS_CHECK(argc, 4);
int workerNum = atoi(argv[3]); // 进程池大小
ProcessData *workerList = (ProcessData *)calloc(workerNum, sizeof(ProcessData));
ERROR_CHECK(workerList, NULL, "calloc");

// 2. 创建工作进程池
makeChild(workerList, workerNum);
printf("进程池创建成功，工作进程数：%d\\n", workerNum);

// 3. TCP初始化（监听网络连接）
int sockFd;
tcpInit(argv[1], argv[2], &sockFd);
printf("服务器启动，监听：%s:%s\\n", argv[1], argv[2]);

// 4. 初始化epoll实例
int epfd;
epollCtor(&epfd);

// 5. 添加监听套接字和工作进程通信套接字到epoll
epollAdd(sockFd, epfd);
for (int i = 0; i < workerNum; i++) {
epollAdd(workerList[i].pipeFd, epfd);
}

// 6. 主进程epoll循环（任务分发）
masterEpollLoop(sockFd, epfd, workerList, workerNum);

// 7. 资源释放（实际运行中需手动终止，此处为规范）
free(workerList);
close(sockFd);
close(epfd);
return 0;
}

3.5 客户端测试代码

// client.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#define ERROR_CHECK(ret, val, msg) { \\
if (ret == val) { \\
perror(msg); \\
exit(1); \\
} \\
}

#define ARGS_CHECK(argc, n) { \\
if (argc != n) { \\
fprintf(stderr, "用法：%s 服务器IP 端口号\\n", argv[0]); \\
exit(1); \\
} \\
}

#define BUF_SIZE 4096

int main(int argc, char *argv[]) {
ARGS_CHECK(argc, 3);

// 1. 创建TCP套接字
int sockFd = socket(AF_INET, SOCK_STREAM, 0);
ERROR_CHECK(sockFd, –1, "socket");

// 2. 配置服务器地址
struct sockaddr_in addr;
bzero(&addr, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = inet_addr(argv[1]);
addr.sin_port = htons(atoi(argv[2]));

// 3. 连接服务器
int ret = connect(sockFd, (struct sockaddr *)&addr, sizeof(addr));
ERROR_CHECK(ret, –1, "connect");
printf("连接服务器 %s:%s 成功\\n", argv[1], argv[2]);

// 4. 输入文件路径并发送
char filePath[128];
printf("请输入要下载的文件路径（如/home/xxx/test.txt）：");
fgets(filePath, sizeof(filePath), stdin);
filePath[strcspn(filePath, "\\n")] = '\\0'; // 去除换行符
send(sockFd, filePath, strlen(filePath), 0);

// 5. 接收文件内容并保存到本地
char buf[BUF_SIZE];
FILE *fp = fopen("downloaded_file", "w");
ERROR_CHECK(fp, NULL, "fopen");

while ((ret = recv(sockFd, buf, sizeof(buf), 0)) > 0) {
fwrite(buf, 1, ret, fp);
bzero(buf, sizeof(buf));
}

printf("文件下载完成，保存为 downloaded_file\\n");

// 6. 关闭资源
fclose(fp);
close(sockFd);
return 0;
}

3.6 编译与运行

1. 编译脚本（Makefile）

# Makefile
CC = gcc
CFLAGS = -Wall -g
TARGET = process_pool
CLIENT_TARGET = client
OBJS = main.o tcpInit.o process_pool.o worker_handle.o master_epoll_loop.o fd_transfer.o epoll_func.o
CLIENT_OBJS = client.o

all: $(TARGET) $(CLIENT_TARGET)

$(TARGET): $(OBJS)
$(CC) $(OBJS) -o $(TARGET)

$(CLIENT_TARGET): $(CLIENT_OBJS)
$(CC) $(CLIENT_OBJS) -o $(CLIENT_TARGET)

%.o: %.c
$(CC) $(CFLAGS) -c $< -o $@

clean:
rm -f $(OBJS) $(TARGET) $(CLIENT_OBJS) $(CLIENT_TARGET) downloaded_file

2. 运行步骤

# 编译
make

# 终端1：启动进程池服务器（IP=192.168.30.129，端口=8080，进程池大小=4）
./process_pool 192.168.30.129 8080 4

# 终端2：启动客户端（连接服务器并下载文件）
./client 192.168.30.129 8080
# 输入文件路径（如/home/xxx/test.txt），下载完成后查看 downloaded_file

四、进程池关键技术细节

4.1 本地套接字（socketpair）

• 特点：全双工通信，效率高于管道（无需处理网络协议），支持传递文件描述符；
• 参数说明：
  • domain=AF_LOCAL：本地协议域（必填）；
  • type=SOCK_STREAM：流式传输（可靠）；
  • sv[2]：存储两个套接字描述符，无读写区分，双向通信。

4.2 文件描述符传递的核心原理

• 传递的不是文件描述符的数值（如3），而是文件描述符对应的内核文件对象；
• 内核会为接收方创建一个新的文件描述符，指向同一个文件对象，实现资源共享；
• 必须使用sendmsg/recvmsg+SCM_RIGHTS类型，普通write/read无法传递。

4.3 epoll IO多路复用的优势

• 主进程通过epoll同时监听“客户端连接”和“工作进程状态反馈”，避免阻塞在单一事件上；
• 支持海量文件描述符监听（无上限，优于select的1024限制）；
• 水平触发（LT）模式，编程简单，适合进程池的状态监控场景。

4.4 工作进程状态管理

• 主进程通过workerList数组维护所有工作进程的PID、状态（FREE/BUSY）和通信套接字；
• 任务分发时采用轮询策略，选择第一个空闲进程，实现简单且负载均衡；
• 工作进程完成任务后，主动发送“FREE”通知，主进程更新状态，实现循环复用。

4.5 异常处理与健壮性

• 工作进程异常退出：主进程通过recv返回0检测，删除对应的epoll监听和文件描述符（可扩展为自动重建工作进程）；
• 客户端断开连接：工作进程通过recv返回0检测，关闭netFd并通知主进程；
• 文件不存在：工作进程向客户端返回错误信息，避免阻塞任务流程；
• 服务器繁忙：无空闲工作进程时，向客户端返回提示，拒绝连接。

五、线程池的核心实现（简化版）

线程池与进程池架构类似，核心差异在于：线程共享进程资源，无需IPC机制，通信成本更低。

5.1 线程池核心结构体

// thread_pool.h
#ifndef __THREAD_POOL_H__
#define __THREAD_POOL_H__

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

// 任务结构体
typedef struct Task {
void (*func)(void *arg); // 任务函数
void *arg; // 任务参数
struct Task *next; // 链表节点（任务队列）
} Task;

// 线程池结构体
typedef struct {
pthread_t *threadIds; // 线程ID数组
int threadNum; // 线程数量
pthread_mutex_t mutex; // 互斥锁（保护任务队列）
pthread_cond_t cond; // 条件变量（唤醒空闲线程）
Task *taskQueue; // 任务队列头节点
Task *taskTail; // 任务队列尾节点
int taskCount; // 当前任务数
int isShutdown; // 是否关闭线程池（0=运行，1=关闭）
} ThreadPool;

// 函数声明
ThreadPool *threadPoolCreate(int threadNum); // 创建线程池
int threadPoolAddTask(ThreadPool *pool, void (*func)(void *), void *arg); // 添加任务
void threadPoolDestroy(ThreadPool *pool); // 销毁线程池
void *threadFunc(void *arg); // 线程工作函数

#endif

5.2 线程池核心实现

// thread_pool.c
#include "thread_pool.h"

// 线程工作函数（循环处理任务）
void *threadFunc(void *arg) {
ThreadPool *pool = (ThreadPool *)arg;

while (1) {
// 加锁保护任务队列
pthread_mutex_lock(&pool->mutex);

// 无任务且未关闭，阻塞等待
while (pool->taskCount == 0 && !pool->isShutdown) {
pthread_cond_wait(&pool->cond, &pool->mutex);
}

// 线程池关闭，解锁并退出
if (pool->isShutdown) {
pthread_mutex_unlock(&pool->mutex);
pthread_exit(NULL);
}

// 取出任务队列头部任务
Task *task = pool->taskQueue;
pool->taskQueue = task->next;
pool->taskCount—;
if (pool->taskCount == 0) {
pool->taskTail = NULL;
}

// 解锁，执行任务
pthread_mutex_unlock(&pool->mutex);
task->func(task->arg);
free(task); // 释放任务内存
}
}

// 创建线程池
ThreadPool *threadPoolCreate(int threadNum) {
ThreadPool *pool = (ThreadPool *)malloc(sizeof(ThreadPool));
if (pool == NULL) return NULL;

// 初始化线程ID数组
pool->threadIds = (pthread_t *)malloc(sizeof(pthread_t) * threadNum);
if (pool->threadIds == NULL) {
free(pool);
return NULL;
}

// 初始化互斥锁和条件变量
pthread_mutex_init(&pool->mutex, NULL);
pthread_cond_init(&pool->cond, NULL);

// 初始化任务队列
pool->taskQueue = NULL;
pool->taskTail = NULL;
pool->taskCount = 0;
pool->isShutdown = 0;
pool->threadNum = threadNum;

// 创建线程
for (int i = 0; i < threadNum; i++) {
if (pthread_create(&pool->threadIds[i], NULL, threadFunc, pool) != 0) {
// 线程创建失败，释放已分配资源
for (int j = 0; j < i; j++) {
pthread_cancel(pool->threadIds[j]);
}
free(pool->threadIds);
pthread_mutex_destroy(&pool->mutex);
pthread_cond_destroy(&pool->cond);
free(pool);
return NULL;
}
}

return pool;
}

// 添加任务到线程池
int threadPoolAddTask(ThreadPool *pool, void (*func)(void *), void *arg) {
if (pool == NULL || func == NULL) return –1;

// 创建任务节点
Task *task = (Task *)malloc(sizeof(Task));
if (task == NULL) return –1;
task->func = func;
task->arg = arg;
task->next = NULL;

// 加锁保护任务队列
pthread_mutex_lock(&pool->mutex);

// 添加任务到队列尾部
if (pool->taskTail == NULL) {
pool->taskQueue = task;
pool->taskTail = task;
} else {
pool->taskTail->next = task;
pool->taskTail = task;
}
pool->taskCount++;

// 唤醒一个空闲线程
pthread_cond_signal(&pool->cond);

// 解锁
pthread_mutex_unlock(&pool->mutex);
return 0;
}

// 销毁线程池
void threadPoolDestroy(ThreadPool *pool) {
if (pool == NULL) return;

// 标记线程池关闭
pthread_mutex_lock(&pool->mutex);
pool->isShutdown = 1;
pthread_mutex_unlock(&pool->mutex);

// 唤醒所有线程
pthread_cond_broadcast(&pool->cond);

// 等待所有线程退出
for (int i = 0; i < pool->threadNum; i++) {
pthread_join(pool->threadIds[i], NULL);
}

// 释放任务队列剩余任务
Task *tmp = NULL;
while (pool->taskQueue != NULL) {
tmp = pool->taskQueue;
pool->taskQueue = tmp->next;
free(tmp);
}

// 释放资源
free(pool->threadIds);
pthread_mutex_destroy(&pool->mutex);
pthread_cond_destroy(&pool->cond);
free(pool);
}

5.3 线程池使用示例（文件下载任务）

// thread_pool_test.c
#include "thread_pool.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#define PORT 8080
#define THREAD_NUM 4
#define BUF_SIZE 4096

// 任务函数：处理客户端文件下载请求
void downloadTask(void *arg) {
int netFd = *(int *)arg;
free(arg); // 释放参数内存

char filePath[128];
bzero(filePath, sizeof(filePath));
int ret = recv(netFd, filePath, sizeof(filePath), 0);
if (ret <= 0) {
close(netFd);
return;
}

printf("线程处理文件下载：%s\\n", filePath);
int fileFd = open(filePath, O_RDONLY);
if (fileFd == –1) {
send(netFd, "文件不存在", strlen("文件不存在"), 0);
close(netFd);
return;
}

char buf[BUF_SIZE];
while ((ret = read(fileFd, buf, sizeof(buf))) > 0) {
send(netFd, buf, ret, 0);
bzero(buf, sizeof(buf));
}

close(fileFd);
close(netFd);
printf("文件下载完成：%s\\n", filePath);
}

int main() {
// 创建线程池
ThreadPool *pool = threadPoolCreate(THREAD_NUM);
if (pool == NULL) {
perror("threadPoolCreate");
exit(1);
}

// TCP初始化
int sockFd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in addr = {AF_INET, htons(PORT), INADDR_ANY};
int reuse = 1;
setsockopt(sockFd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
bind(sockFd, (struct sockaddr *)&addr, sizeof(addr));
listen(sockFd, 10);
printf("线程池服务器启动，监听端口%d\\n", PORT);

// 接收客户端连接，添加任务到线程池
struct sockaddr_in clientAddr;
socklen_t clientLen = sizeof(clientAddr);
while (1) {
int *netFd = (int *)malloc(sizeof(int));
*netFd = accept(sockFd, (struct sockaddr *)&clientAddr, &clientLen);
if (*netFd == –1) continue;

printf("客户端连接：%s:%d\\n",
inet_ntoa(clientAddr.sin_addr), ntohs(clientAddr.sin_port));
threadPoolAddTask(pool, downloadTask, netFd);
}

// 销毁线程池（实际运行中不会执行）
threadPoolDestroy(pool);
close(sockFd);
return 0;
}

六、核心总结与进阶方向

6.1 核心知识点总结

6.2 进阶学习方向