Linux 进程池原理与实现：从核心逻辑到代码实战

这里有个细节要注意：用 std::function<void()> 定义任务类型，实现'任务解耦'，后续可轻松添加新任务，无需修改进程池核心逻辑。任务码（0-3）对应不同任务，通过 RandomTask 模拟真实场景中随机到来的任务请求。

2. 实现子进程工作逻辑

子进程的核心作用是'等待任务、执行任务'，通过读取管道中的任务码，调用对应的任务函数，这是进程复用的核心逻辑。

// 子进程工作函数：从管道读取任务码，执行对应任务
void Work(int rfd) {
    while (true) {
        int code = 0;
        // 从管道读任务码（阻塞等待，直到主进程发送任务）
        ssize_t n = read(rfd, &code, sizeof(code));
        if(n == sizeof(int)) { // 读取到完整任务码
            // 任务码合法，执行对应任务
            if(code >= 0 && code < gtasks.size()) {
                gtasks[code](); // 调用任务容器中的对应任务
            }
        } else if(n == 0) { // 读取到 EOF（主进程关闭管道写端），子进程退出
            break;
        } else { // 读取失败，子进程退出
            break;
        }
    }
}

子进程通过 read 从管道读端获取任务码，阻塞等待任务，实现'空闲时待命、有任务时执行'。只有读取到完整的任务码（4 字节，int 类型），才执行任务，避免任务错乱。当主进程关闭管道写端，子进程读取到 n=0，主动退出，避免僵尸进程。

3. 封装 Channel 类——管理通信与子进程

Channel 类是进程池的'通信桥梁'，封装了主进程与单个子进程的管道写端、子进程 PID，统一管理任务发送、管道关闭、子进程回收，简化进程池的管理逻辑。

class Channel {
public:
    // 构造函数：初始化管道写端和子进程 PID
    Channel(int wfd, pid_t who) : _wfd(wfd), _sub_process_id(who) {
        _name = "Channel-" + std::to_string(_sub_process_id) + "-" + std::to_string(_wfd);
    }

    // 获取管道写端（供主进程发送任务使用）
    int Fd() { return _wfd; }

    // 获取子进程 PID（供回收子进程使用）
    pid_t SubId() { return _sub_process_id; }

    // 获取 Channel 名称（便于日志打印和调试）
    std::string Name() { return _name; }

    // 关闭管道写端（通知子进程退出）
    void Close() {
        if(_wfd >= 0) close(_wfd);
    }

    // 回收子进程（避免僵尸进程）
    void Wait() {
        pid_t rid = waitpid(_sub_process_id, nullptr, 0);
        (void)rid; // 消除未使用变量警告
    }

    // 向子进程发送任务（写入任务码到管道）
    void SendTask(int taskcode) {
        ssize_t n = write(_wfd, &taskcode, sizeof(taskcode));
        (void)n; // 简化处理，实际可增加错误判断
    }

    ~Channel() {}

private:
    int _wfd;             // 管道写端（主进程用，向子进程发任务）
    pid_t _sub_process_id; // 对应的子进程 PID
    std::string _name;     // Channel 名称（用于调试和日志）
};

每个子进程对应一个 Channel 对象，主进程通过 Channel 的 SendTask 发送任务，通过 Close 和 Wait 回收子进程。封装后，进程池无需直接操作管道和子进程 PID，降低耦合度。

4. 封装 ProcessPool 类——核心管理逻辑

ProcessPool 类是整个进程池的'管理者'，负责创建子进程、管理所有 Channel、实现负载均衡分配任务、终止进程池，是面向对象设计的核心。

class ProcessPool {
private:
    // 负载均衡：轮询选择子进程（Next 函数实现轮询逻辑）
    int Next() {
        int choice = _next_choice;
        _next_choice++;
        _next_choice %= _channels.size(); // 取模实现循环轮询
        return choice;
    }

public:
    // 构造函数：初始化进程池大小（子进程数量）
    ProcessPool(int number) :_number(number), _next_choice(0) {}

    // 启动进程池：创建指定数量的子进程和对应 Channel
    void Start() {
        for (int i = 0; i < _number; i++) {
            // 1. 创建管道（主写子读）
            int pipefd[2];
            int n = pipe(pipefd);
            if (n < 0) {
                perror("pipe:");
                exit(2);
            }

            // 2. 创建子进程
            pid_t id = fork();
            if (id < 0) {
                perror("fork:");
                exit(3);
            } else if (id == 0) { // 子进程逻辑
                close(pipefd[1]); // 子进程关闭写端，只保留读端
                Work(pipefd[0]);  // 子进程进入工作循环，等待任务
                close(pipefd[0]); // 执行完任务，关闭读端
                exit(0);          // 子进程退出
            } else { // 主进程逻辑
                close(pipefd[0]); // 主进程关闭读端，只保留写端
                // 创建 Channel 对象，管理当前子进程和管道写端，存入容器
                _channels.emplace_back(pipefd[1], id);
            }
        }
    }

    // 推送任务：根据轮询策略，将任务发送给子进程
    void PushTask(int taskcode) {
        int who = Next(); // 轮询选择一个子进程（负载均衡）
        _channels[who].SendTask(taskcode); // 发送任务码
        // 打印日志，直观查看任务分配情况
        std::cout << "发送任务：" << Task2String(taskcode) << "[" << taskcode << "]" << "给：" << _channels[who].Name() << std::endl;
    }

    // 终止进程池：关闭所有管道写端，回收所有子进程
    void Stop() {
        // 从后往前关闭管道、回收子进程（避免容器迭代时出现异常）
        int end = _channels.size() - 1;
        while(end >= 0) {
            _channels[end].Close(); // 关闭管道写端，通知子进程退出
            _channels[end].Wait();  // 回收子进程
            std::cout << _channels[end].Name() << " close and wait success!" << std::endl;
            end--;
        }
    }

    // 调试打印：输出所有 Channel 的信息
    void DebugPrint() {
        std::cout << "--------------------------------" << std::endl;
        for(auto &channel : _channels) {
            std::cout << channel.Fd() << std::endl;
            std::cout << channel.SubId() << std::endl;
            std::cout << channel.Name() << std::endl;
        }
        std::cout << "--------------------------------" << std::endl;
    }

    ~ProcessPool() {}

private:
    std::vector<Channel> _channels; // 存储所有 Channel，管理子进程和通信
    int _number;                    // 进程池大小（子进程数量）
    int _next_choice;               // 轮询索引，实现负载均衡
};

Start() 函数批量创建子进程和管道，主进程关闭管道读端、保存写端到 Channel，子进程关闭写端、进入 Work 循环等待任务。Next() 函数通过轮询分配任务，确保所有子进程负载均匀。Stop() 函数从后往前关闭管道写端、回收子进程，确保所有资源被正确回收。

5. 主函数测试

主函数实现进程池的完整调用流程：解析命令行参数、加载任务、生成随机任务、启动进程池、推送任务、终止进程池。

#ifdef __MAIN__
// 用法提示：输入进程池大小（如./process_pool 5）
static void Usage(const std::string &proc) {
    std::cout << "Usage:\n\t" << proc << " process_number" << std::endl;
}

// 程序入口：./process_pool 进程池大小
int main(int argc, char *argv[]) {
    // 1. 解析命令行参数（必须输入进程池大小）
    if (argc != 2) {
        Usage(argv[0]);
        exit(1);
    }
    int number = std::stoi(argv[1]); // 进程池大小（子进程数量）

    // 2. 初始化任务：加载所有测试任务，生成 50 个随机任务
    srand(time(nullptr)^getpid()); // 设置随机种子（结合时间和 PID，确保随机性）
    LoadTask();
    std::vector<int> task_codes;
    RandomTask(&task_codes);

    // 3. 创建进程池（智能指针管理，自动释放，避免内存泄漏）
    std::unique_ptr<ProcessPool> pp = std::make_unique<ProcessPool>(number);

    // 4. 启动进程池，创建子进程
    pp->Start();
    sleep(2); // 等待子进程初始化完成

    // 5. 推送所有随机任务
    for(auto task : task_codes) {
        pp->PushTask(task);
        usleep(500000); // 模拟任务推送间隔，避免任务堆积
    }

    // 6. 终止进程池，回收所有资源
    pp->Stop();
    return 0;
}
#endif

用 std::unique_ptr 管理 ProcessPool 对象，自动释放内存，避免手动管理内存导致的泄漏。命令行参数解析：输入进程池大小（如 ./process_pool 5，表示创建 5 个子进程），符合 Linux 程序的使用习惯。

三、编译运行与结果分析

Makefile 内容如下：

process_pool: process_pool.cc
	g++ -o $@ $^ -std=c++14
.PHONY: clean
clean:
	rm -f process_pool

• 编译：直接 make；
• 运行：./process_pool 5（5 为子进程数量，可自定义）；

结果分析：

• 5 个子进程循环复用，处理 50 个任务，无需频繁创建/销毁子进程；
• 销毁进程池时，所有子进程正常退出，资源被回收，无僵尸进程。

四、完整代码展示

为了方便大家直接测试和修改，以下是整合后的完整代码：

#include <iostream>
#include <vector>
#include <string>
#include <functional>
#include <ctime>
#include <cstdlib>
#include <unistd.h>
#include <memory>
#include <sys/wait.h>

#define __MAIN__

//////////////////////////////任务测试代码/////////////////////////////
using task_t = std::function<void()>;

void printlog() {
    // sleep(1);
    std::cout << "我是一个打印日志的任务，pid: " << getpid() << std::endl;
}

void download() {
    // sleep(1);
    std::cout << "我是一个下载任务，pid: " << getpid() << std::endl;
}

void readmysql() {
    // sleep(1);
    std::cout << "我是一个访问数据库的操作，pid: " << getpid() << std::endl;
}

void writeredis() {
    // sleep(1);
    std::cout << "我是一个访问 redis 的任务，pid: " << getpid() << std::endl;
}

std::vector<task_t> gtasks;

void LoadTask() {
    gtasks.push_back(printlog);
    gtasks.push_back(download);
    gtasks.push_back(readmysql);
    gtasks.push_back(writeredis);
}

void RandomTask(std::vector<int> *out) {
    for(int i=0; i<50; i++) {
        int code = rand() % gtasks.size();
        usleep(23223);
        out->push_back(code);
    }
}

#define LOG_TASK 0
#define DOWNLOAD_TASK 1
#define MYSQL_TASK 2
#define REDIS_TASK 3

std::string Task2String(int code) {
    switch(code) {
        case LOG_TASK: return "printlog";
        case DOWNLOAD_TASK: return "download";
        case MYSQL_TASK: return "readmysql";
        case REDIS_TASK: return "writeredis";
        default: return "unknown";
    }
}

//////////////////////////////进程池代码/////////////////////////////
void Work(int rfd) {
    while (true) {
        int code = 0;
        ssize_t n = read(rfd, &code, sizeof(code));
        if(n == sizeof(int)) {
            if(code >= 0 && code < gtasks.size()) {
                gtasks[code]();
            }
        } else if(n == 0) {
            break; // 子进程只要读到返回值为 0，表明父进程让我退出
        } else {
            break;
        }
    }
}

class Channel {
public:
    Channel(int wfd, pid_t who) : _wfd(wfd), _sub_process_id(who) {
        _name = "Channel-" + std::to_string(_sub_process_id) + '-' + std::to_string(_wfd);
    }
    int Fd() { return _wfd; }
    pid_t SubId() { return _sub_process_id; }
    std::string Name() { return _name; }
    void Close() { if(_wfd >= 0) close(_wfd); }
    void Wait() { pid_t rid = waitpid(_sub_process_id, nullptr, 0); (void)rid; }
    void SendTask(int taskcode) { ssize_t n = write(_wfd, &taskcode, sizeof(taskcode)); (void)n; }
    ~Channel() {}
private:
    int _wfd;
    pid_t _sub_process_id;
    std::string _name;
};

class ProcessPool {
private:
    int Next() {
        int choice = _next_choice;
        _next_choice++;
        _next_choice %= _channels.size();
        return choice;
    }
public:
    ProcessPool(int number) :_number(number), _next_choice(0) {}

    void Start() {
        for (int i = 0; i < _number; i++) {
            int pipefd[2];
            int n = pipe(pipefd);
            if (n < 0) { perror("pipe:"); exit(2); }
            pid_t id = fork();
            if (id < 0) { perror("fork:"); exit(3); }
            else if (id == 0) { // 子进程
                close(pipefd[1]);
                Work(pipefd[0]);
                close(pipefd[0]);
                exit(0);
            } else { // 父进程
                close(pipefd[0]);
                _channels.emplace_back(pipefd[1], id);
            }
        }
    }

    void PushTask(int taskcode) {
        int who = Next();
        _channels[who].SendTask(taskcode);
        std::cout << "发送任务：" << Task2String(taskcode) << "[" << taskcode << "]" << "给：" << _channels[who].Name() << std::endl;
    }

    void Stop() {
        int end = _channels.size() - 1;
        while(end >= 0) {
            _channels[end].Close();
            _channels[end].Wait();
            std::cout << _channels[end].Name() << "close and wait success!" << std::endl;
            end--;
        }
    }

    void DebugPrint() {
        std::cout << "--------------------------------" << std::endl;
        for(auto &channel : _channels) {
            std::cout << channel.Fd() << std::endl;
            std::cout << channel.SubId() << std::endl;
            std::cout << channel.Name() << std::endl;
        }
        std::cout << "--------------------------------" << std::endl;
    }
    ~ProcessPool() {}
private:
    std::vector<Channel> _channels;
    int _number;
    int _next_choice;
};

#ifdef __MAIN__
static void Usage(const std::string &proc) {
    std::cout << "Usage:\n\t" << proc << " process_number" << std::endl;
}

int main(int argc, char *argv[]) {
    if (argc != 2) {
        Usage(argv[0]);
        exit(1);
    }
    int number = std::stoi(argv[1]);
    srand(time(nullptr)^getpid());
    LoadTask();
    std::vector<int> task_codes;
    RandomTask(&task_codes);

    std::unique_ptr<ProcessPool> pp = std::make_unique<ProcessPool>(number);
    pp->Start();
    sleep(2);

    for(auto task : task_codes) {
        pp->PushTask(task);
        usleep(500000);
    }
    pp->Stop();
    return 0;
}
#endif

五、进阶优化：让进程池更实用

我们手搓的是基础版进程池，实际生产中可根据需求优化以下几点：

1. 动态调整进程池大小：根据任务量自动增加/减少子进程（避免空闲子进程浪费资源）；
2. 任务类型扩展：支持传入函数指针和参数，让进程池能处理不同类型的任务（而非固定打印）；
3. 非阻塞添加任务：任务队列满时，返回错误而非阻塞，提高主进程灵活性；
4. 信号处理优化：处理 SIGCHLD 信号，及时回收异常退出的子进程，并重新创建子进程，保证进程池稳定性；
5. 使用更高效的 IPC：将管道替换为消息队列（支持消息优先级）或共享内存（更高吞吐量）。

六、常见坑点与注意事项

• 管道读写阻塞：子进程 read() 会阻塞，主进程 write() 在管道满时也会阻塞，需根据需求调整为非阻塞；
• 僵尸进程：必须用 waitpid() 回收子进程，尤其是子进程异常退出时，避免僵尸进程占用资源；
• 管道关闭顺序：主进程需先关闭写端，子进程才能读取到 EOF 并正常退出；
• 内存泄漏：进程池销毁时，必须释放所有分配的内存；
• 任务队列溢出：需设置合理的 max_task，避免任务堆积导致内存溢出。

七、总结

通过手搓这个基础版 Linux 进程池，我们搞懂了进程池的核心逻辑——提前创建子进程、复用子进程、统一管理任务，本质上是用空间换时间，减少进程创建/销毁的开销。

本文的代码虽然简单，但涵盖了进程池的核心流程，你可以在此基础上进行扩展，适配实际业务场景（如 Web 服务器的请求处理、后台任务调度等）。

如果运行过程中遇到问题，可重点检查管道操作、子进程回收和任务队列的逻辑，这些都是实现进程池的关键。动手敲一遍代码，你会对 Linux 进程控制和 IPC 有更深刻的理解～