基于管道通信的 Linux 进程池实现详解

子进程任务处理逻辑

Work 函数是子进程的核心执行逻辑，负责从管道读取任务码并执行对应任务。这里需要注意 read 的返回值判断，当返回 0 时通常表示对端关闭了写端。

// 子进程工作函数：循环读取管道中的任务码并执行
// rfd：管道读端文件描述符
void Work(int rfd) {
    while (true) {
        int code = 0;
        // 从管道读取任务码（阻塞读）
        ssize_t n = read(rfd, &code, sizeof(code));
        
        // 读取成功且长度正确：执行任务
        if (n > 0 && n == sizeof(int)) {
            if (code >= 0 && code < gtasks.size()) {
                gtasks[code](); // 执行对应任务
            }
        }
        // 读取到 0：表示管道写端关闭（父进程通知退出）
        else if (n == 0) {
            break; // 子进程退出循环
        }
        // 读取错误：直接退出
        else {
            break;
        }
    }
}

通道（Channel）类：封装父子进程通信

Channel 类封装了'管道写端 + 子进程 ID'的关联关系，简化父进程对单个子进程的管理（发任务、关管道、回收进程）。

class Channel {
public:
    // 构造函数：初始化管道写端、子进程 ID
    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; // 实际场景应检查写操作是否成功
    }

private:
    int _wfd;              // 管道写端文件描述符
    pid_t _sub_process_id; // 对应子进程 ID
    std::string _name;     // 通道名称（调试用）
};

进程池（ProcessPool）类：核心管理逻辑

ProcessPool 类是进程池的核心，负责创建子进程、管理通道、分发任务、停止进程池。注意这里修正了原代码中的拼写错误 ProcesspPool -> ProcessPool。

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) {
        std::cout << "number: " << _number << std::endl;
    }

    // 启动进程池（父进程执行）：创建指定数量的子进程和管道
    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]); // 父进程关闭读端（只写）
                // 创建通道对象并加入管理列表
                _channels.emplace_back(pipefd[1], id);
            }
        }
    }

    // 推送任务：选择子进程并发送任务码
    void PushTask(int taskcode) {
        int who = Next();
        _channels[who].SendTask(taskcode);
        std::cout << "发送任务：" << TaskToString(taskcode) << "[" << taskcode << "]" 
                  << "给：" << _channels[who].Name() << std::endl;
    }

    // 停止进程池：关闭所有管道，回收子进程
    void Stop() {
        // Version1 -- 可以成功
        // 1. 批量关闭所有管道写端（通知子进程退出）
        for (auto& channel : _channels) {
            channel.Close();
            std::cout << channel.Name() << " close success!" << std::endl;
        }
        sleep(3); // 等待子进程处理完最后任务并退出

        // 2. 批量回收子进程资源
        for (auto& channel : _channels) {
            channel.Wait();
            std::cout << channel.Name() << " wait success!" << std::endl;
        }
        
        // Version2 -- 不能成功（原因：关闭管道后立即 wait，子进程可能还在阻塞读）
        // for(auto& channel : _channels) {
        //     channel.Close();
        //     channel.Wait();
        // }
    }

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 << " proceess_number" << std::endl;
}

int main(int argc, char* argv[]) {
    if (argc != 2) {
        Usage(argv[0]);
        exit(1);
    }

    int number = std::stoi(argv[1]);

    // 0. 初始化：加载任务、随机生成 50 个任务码
    srand(time(nullptr) ^ getpid());
    LoadTask();
    std::vector<int> task_codes;
    RandomTask(&task_codes);

    // 1. 创建进程池对象
    std::unique_ptr<ProcessPool> pp = std::make_unique<ProcessPool>(number);

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

    // 3. 分发所有随机任务
    for (auto task : task_codes) {
        pp->PushTask(task);
        usleep(500000); // 模拟任务分发间隔（500ms）
    }

    // 4. 停止进程池（回收资源）
    pp->Stop();
    return 0;
}
#endif

关键知识点解析

1. 管道通信原理

匿名管道 pipe() 创建的文件描述符对 pipefd[0]（读）、pipefd[1]（写）是单向的。父子进程继承管道文件描述符，通过关闭不需要的端实现'父写子读'。当写端关闭后，读端 read() 会返回 0，子进程通过这个信号判断退出。

2. 轮询负载均衡

Next() 函数通过递增取模的方式，循环选择子进程，确保任务均匀分发给所有子进程，避免单个子进程过载。

3. 进程回收的坑

在 Stop() 函数中，如果采用'关闭一个管道立即回收一个子进程'的策略（Version2），可能会导致死锁。因为父进程在等待一个子进程时，其他子进程可能仍在阻塞读取管道，而父进程尚未关闭它们的写端。

正确的做法是 Version1：先批量关闭所有管道写端，让所有子进程感知到退出信号并结束循环，然后再统一调用 waitpid 回收资源。这样避免了资源竞争导致的阻塞。

编译与运行

项目包含一个 Makefile，配置如下：

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

• 编译：直接执行 make；
• 运行：./process_pool 5（5 为子进程数量，可自定义）；
• 输出：可以看到任务被轮询分发给不同子进程，每个任务打印对应的进程 ID，最后进程池正常停止并回收资源。

扩展与优化方向

虽然这个实现已经跑通了基本流程，但在生产环境中还可以考虑以下优化：

• 错误处理：当前代码未处理 write()/read() 的错误返回值，实际场景应增加重试、日志记录；
• 动态扩容：支持运行时增加 / 减少子进程数量；
• 更优的负载均衡：基于子进程当前任务数、CPU 使用率等动态分发；
• 任务队列：父进程增加任务队列，避免任务分发过快导致管道阻塞；
• 信号处理：增加 SIGCHLD 信号处理，异步回收子进程，避免僵尸进程。

通过拆解这个极简的进程池实现，我们可以深入理解 Linux 进程间通信、进程管理的核心知识点。这个案例涵盖了进程池的核心设计思想，是学习 Linux 并发编程的绝佳实践。