Linux 进程池实战：基于管道通信的任务分发系统实现

2. 子进程工作逻辑

子进程的核心是 Work 函数。它循环从管道读取任务码，执行对应任务。当读取到 0（表示管道写端关闭）时，子进程退出。

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](); // 执行任务
            }
        } else if (n == 0) {
            break; // 管道写端关闭，退出
        } else {
            break; // 读取错误，退出
        }
    }
}

3. 通道类封装

Channel 类封装了单个子进程的通信管道写端和进程 ID，简化父进程的管理操作（发送任务、关闭连接、等待回收）。

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;
    }

private:
    int _wfd;
    pid_t _sub_process_id;
    std::string _name;
};

4. 进程池管理核心

ProcessPool 类负责创建子进程、维护通道列表、分发任务和停止池。这里采用了轮询算法 (Next) 来实现简单的负载均衡。

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 << "Creating pool with " << _number << " workers." << std::endl;
    }

    void Start() {
        for (int i = 0; i < _number; i++) {
            int pipefd[2];
            if (pipe(pipefd) < 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 << "Sent task: " << TaskToString(taskcode) << " to " << _channels[who].Name() << std::endl;
    }

    void Stop() {
        // 策略：先批量关闭所有写端，再统一回收
        for (auto& channel : _channels) {
            channel.Close();
            std::cout << channel.Name() << " closed." << std::endl;
        }
        sleep(3); // 等待子进程处理完剩余任务

        for (auto& channel : _channels) {
            channel.Wait();
            std::cout << channel.Name() << " reaped." << std::endl;
        }
    }

private:
    std::vector<Channel> _channels;
    int _number;
    int _next_choice;
};

5. 主函数入口

主函数负责初始化环境、启动进程池、分发任务并最终停止。

#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);

    auto 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. 管道通信与退出信号

匿名管道 pipe() 创建的读端和写端是单向的。父子进程继承文件描述符后，父进程关闭读端，子进程关闭写端。关键点在于：当管道的所有写端都关闭时，读端的 read() 会返回 0。这是子进程感知父进程要求退出的主要信号。

2. 优雅退出的陷阱

在 Stop() 方法中，如果采用'关闭一个管道后立即等待该进程'的策略（即串行关闭回收），可能会导致死锁或资源竞争。因为父进程在等待某个子进程退出时，可能无法及时处理其他子进程的状态，或者子进程在等待写入时父进程已阻塞。

推荐做法：先遍历关闭所有管道的写端，给子进程足够的时间处理完当前任务并检测到 EOF，然后再统一调用 waitpid 回收资源。这样能确保所有子进程都能收到退出信号并正常退出。

编译与运行

使用 Makefile 进行编译：

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

编译命令：make 运行命令：./process_pool 5（启动 5 个子进程）

扩展与优化方向

• 错误处理：增加对 write/read 返回值的具体检查，记录异常日志。
• 动态扩容：支持运行时调整子进程数量。
• 更优调度：根据子进程负载情况动态分配任务，而非简单轮询。
• 任务队列：父进程增加缓冲队列，防止任务产生过快导致管道阻塞。
• 信号处理：利用 SIGCHLD 异步回收僵尸进程，提高系统稳定性。