C++实现进程池：主从架构、管道通信与任务调度(示例详解)

主/从进程池架构

在现代高并发系统的设计中，进程池是一个非常经典且高效的模型。它的核心思想说起来并不复杂：用有限数量的子进程，去处理源源不断到来的任务，从而避免频繁的进程创建和销毁开销。这里要聊的，就是基于这种思路的一个典型实现——主/从进程池架构。

一、原理

1 个主进程（Master）+ 预先创建 N 个子进程（Worker），这就是整个架构的基础。主进程的角色更像是一个调度中心，它不去处理具体的业务，只管把任务分配下去。而子进程，就是真正干活的工人，它们提前创建好，阻塞等待任务，处理完一个再等着下一个。它们之间通过管道来通信。

这么设计的最大好处是什么？高并发、稳定性强，而且避免了反复创建和销毁进程带来的资源损耗。

二、三大核心角色

1. Master（主进程）

这个角色是唯一的，它的职责非常纯粹——不处理任何业务，只做四件事：

• 创建进程池：通过 fork 系统调用，一次性创建好一批子进程。
• 接收任务：从外部或内部获取待处理的任务。
• 分发任务：将任务分配给当前空闲的子进程。
• 管理子进程：监控子进程的运行状态，必要时重启、回收。

2. Worker（子进程 / 进程池）

这些子进程是预先创建好 N 个的，它们一直活着，不会随意退出。它们的运行逻辑可以简单理解为：阻塞等待任务、收到任务、处理任务、继续等待。整个过程是一个循环。

3. 管道（通信通道）

实现父子进程通信的关键。每个子进程都有一条专属管道。主进程往管道的写端发任务，子进程从管道的读端收任务。由于管道读端默认是阻塞的，这天然就实现了一个同步控制的效果——有任务就处理，没任务就等着。

三、代码

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#define POLL_SIZE 4
// ***********************************任务列表*********************************************
void SyncDisk() {
    std::cout << "同步磁盘..." << std::endl;
}
void DownloadFile() {
    std::cout << "下载文件..." << std::endl;
}
void PrintMessage() {
    std::cout << "打印消息..." << std::endl;
}
void UpdateDatabase() {
    std::cout << "更新数据库..." << std::endl;
}
typedef void (*task_t)();   // 定义一个函数指针类型，指向任务函数
task_t tasks[] = {SyncDisk, DownloadFile, PrintMessage, UpdateDatabase};  // 任务列表
// ***********************************进程池实现*******************************************
enum {
    OK = 0,
    PIPE_ERROR,
    FORK_ERROR
};
void Dotask(int fd) {   // 任务的入口
    while (1) {
        int task_code = 0;
        ssize_t n = read(fd, &task_code, sizeof(task_code));    // 约定好读取4字节的任务码
        if (n == -1) {
            std::cerr << "read error" << std::endl;
            break;
        }
        else if (n == 0) {
            std::cout << "没有任务了..." << std::endl;
            break; // 管道被关闭，退出循环
        }
        else {
            if (task_code < 0 || task_code >= sizeof(tasks) / sizeof(task_t)) {
                std::cerr << "Invalid task code: " << task_code << std::endl;
                continue; // 跳过无效的任务码
            }
            // 根据任务码执行对应的任务函数
            tasks[task_code]();
        }
    }
}
// typedef std::function cb_t;
using cb_t = std::function;
class ProcessPool {
private:
    // 定义一个内部Channel类，保存管道写端和子进程pid
    class Channel {
    public:
        Channel(int wfd, pid_t sub_pid)
            : _wfd(wfd), _sub_pid(sub_pid)
        {
            _sub_name = "sub_channel_" + std::to_string(sub_pid);
        }
        ~Channel()
        {
        }
        void Write(int task_index) {
            ssize_t n = write(_wfd, &task_index, sizeof(task_index));
            (void)n; // 忽略写入结果
        }
        void PrintInfo() {
            printf("Channel Info - Sub PID: %d, Sub Name: %sn", _sub_pid, _sub_name.c_str());
        }
        std::string GetSubName() const {
            return _sub_name;
        }
        void ClosePipe() {
            std::cout << "关闭管道wfd: " << _wfd << std::endl;
            close(_wfd);
        }
        void Wait() {
            waitpid(_sub_pid, nullptr, 0);
            std::cout << "回收子进程: " << " PID: " << _sub_pid << std::endl;
        }
    private:
        int _wfd;              // 写管道文件描述符
        pid_t _sub_pid;        // 子进程pid
        std::string _sub_name; // 子进程名字
    };
public:
    ProcessPool() {
        srand((unsigned int)time(nullptr) ^ (unsigned int)getpid()); // 设置随机数种子
    }
    ~ProcessPool(){}
    void Init(cb_t cb) {
        CreatProcessChannel(cb);
    }
    void Debug() {
        for (auto& ch : channels) {
            ch.PrintInfo();
        }
    }
    void Run() {
        int cnt = 6;
        while (cnt--) {
            std::cout << "---------------------------------------------" << std::endl;
            // 1. 选择一个任务
            int itask = SelectTask();
            std::cout << "选择任务： index: " << itask << std::endl;
            // 2. 选择一个channel(管道+子进程)，本质是选择一个下标
            int index = SelectChannel();
            std::cout << "选择管道： index: " << index << std::endl;
            // 3. 发送任务给指定的channel(管道+子进程)
            SendTaskToChannel(itask, index);
            std::cout << "发送任务 index " << itask << " 到管道 index " << index << std::endl;
        }
    }
    void Quit() {
        // 想要1 : 1地回收，就要改变CreateProcessChannel的实现
        // 就要在子进程中关闭历史遗留的管道写端
        // 从而实现1 : 1地回收资源。
        for (auto& ch : channels) {
            ch.ClosePipe(); // 关闭写端
            ch.Wait();
        }
        // 逆向回收资源
        // int end = channels.size() - 1;
        // for (int i = end; i >= 0; --i) {
        //     channels[i].ClosePipe(); // 关闭写端
        //     channels[i].Wait();      // 等待子进程退出
        // }
        // 1:1回收演示
        // for (auto& ch : channels) {
        //     ch.ClosePipe(); // 关闭写端
        //     ch.Wait();
        // }
        // // 1. 关闭所有管道写端，通知子进程退出
        // for (auto& ch : channels) {
        //     ch.ClosePipe(); // 关闭写端
        // }
        // // 2. 等待所有子进程退出
        // for (auto& ch : channels) {
        //     ch.Wait();
        // }
    }
private:
    int SelectChannel() {
        // 这里简单的轮询选择一个channel
        static int index = 0;
        int selected = index % channels.size();
        ++index;
        return selected;
    }
    int SelectTask() {
        // 轮询选择一个任务
        int itask = rand() % (sizeof(tasks) / sizeof(task_t));
        return itask;
    }
    void SendTaskToChannel(int itask, int index) {
        if (itask < 0 || itask >= sizeof(tasks) / sizeof(task_t)) {
            std::cerr << "Invalid task index: " << itask << std::endl;
            return;
        }
        if (index < 0 || index >= channels.size()) {
            std::cerr << "Invalid channel index: " << index << std::endl;
            return;
        }
        // 将任务索引写入管道，通知子进程执行对应的任务
        channels[index].Write(itask);
    }
    void CreatProcessChannel(cb_t cb) {
        for (int i = 0; i < POLL_SIZE; ++i) {
            int pipefd[2] = {0};
            int ret = pipe(pipefd);
            if (ret == -1) {
                std::cerr << "pipe error" << std::endl;
                exit(PIPE_ERROR);
            }
            pid_t pid = fork();
            if (pid == -1) {
                std::cerr << "fork error" << std::endl;
                exit(FORK_ERROR);
            }
            else if (pid == 0) {
                // 子进程
                if (!channels.empty()) {
                    for (auto& ch : channels) {
                        ch.ClosePipe(); // 关闭历史遗留的管道写端
                    }
                }
                close(pipefd[1]);   // 关闭写端
                cb(pipefd[0]);
                exit(OK);
            }
            else {
                // 父进程
                close(pipefd[0]);   // 关闭读端
                // 创建一个channel对象，保存管道写端和子进程pid
                channels.emplace_back(pipefd[1], pid);
                // Channel ch(pipefd[1], pid);
                // channels.emplace_back(ch);  // 将channel对象添加到容器中
                std::cout << "创建了一个管道PID: " << pid << "到管道容器中了" << std::endl;
                sleep(1);
            }
        }
    }
private:
    std::vector channels;    // 要有未来组织所有channel的容器
};
int main() {
    // 1. 初始化一个进程池对象
    ProcessPool pool;
    pool.Init(Dotask);
    // 2. 运行进程池
    pool.Run();
    // 3. 结束进程池释放资源
    pool.Quit();
    return 0;
}

Linux 进程池（ProcessPool）源码解析

这部分我们从源码角度，深入拆解一下这个进程池的实现。目标很清楚：一个父进程调度任务，多个子进程执行任务，父子之间用匿名管道通信——父进程发任务码，子进程根据码值执行对应函数。

一、项目目标

实现一个简单的进程池：

• 父进程负责调度任务
• 子进程负责执行任务
• 父子进程通过匿名管道通信
• 父进程发送任务码
• 子进程根据任务码执行对应任务

二、整体架构

                     Parent(ProcessPool)
                              │
         ┌────────────────────┼────────────────────┐
         │                    │                    │
         │                    │                    │
      pipe0                pipe1                pipe2
         │                    │                    │
         ▼                    ▼                    ▼
      Child0               Child1               Child2
         │                    │                    │
      Dotask()             Dotask()             Dotask()
         │                    │                    │
         └────────────执行任务────────────┘

父进程即 ProcessPool 对象。它的核心工作就是前面说的：创建子进程、创建管道、选择任务、分发任务、回收子进程。子进程呢？只管执行任务。

三、任务系统

任务定义

void SyncDisk();
void DownloadFile();
void PrintMessage();
void UpdateDatabase();

任务表

typedef void (*task_t)();
task_t tasks[] =
{
    SyncDisk,
    DownloadFile,
    PrintMessage,
    UpdateDatabase
};

本质就是一个映射：任务码 → 函数地址。

四、ProcessPool类设计

class ProcessPool
{
private:
    vector channels;
};

这个类内部用一个 vector 保存了所有的 Channel 对象。每个 Channel 本质上就绑定了一组关系：一个管道写端 + 一个子进程 PID。

五、Channel设计

Channel是什么

可以把它理解为一个“通信单元”，包含：

一个管道 + 一个子进程

类结构

class Channel
{
private:
    int _wfd;          // 管道写端，父进程通过它发送任务
    pid_t _sub_pid;    // 子进程PID，用于waitpid回收
    string _sub_name;  // 调试用的名字，比如 "sub_channel_1234"
};

六、进程创建流程

CreateProcessChannel()

这个函数的核心就两行代码：

pipe(pipefd);
fork();

然后根据 fork 的返回值，分别处理父进程和子进程逻辑。每次循环创建一个子进程和一条管道，子进程关闭管道写端，父进程关闭管道读端。

整个创建过程是逐步构建的：第一次循环生成 Child0 和 pipe0；第二次循环生成 Child1 和 pipe1，此时父进程、Child0、Child1 都共享了 pipe0 的文件描述符；第三次循环同理。最终会形成：

Parent
│
├── Child0
├── Child1
├── Child2
└── Child3

七、为什么关闭历史遗留写端

这是一个非常关键且容易踩坑的地方。由于 fork 会继承所有打开的文件描述符，假设我们创建了 Child0，它持有 pipe0[0] 和 pipe0[1]（虽然子进程关闭了 pipe0[1]，但父进程的 pipe0[1] 依然开着）。当继续 fork Child1 时，Child1 同样会继承父进程的所有文件描述符，其中包括 pipe0[1]。

这就导致了一个问题：父进程关闭了 pipe0 的写端，但 Child1 手上还捏着 pipe0 的写端，那么 Child0 的 read 就无法收到 EOF。所以，子进程在创建后，必须关闭所有之前创建的、属于其他子进程的管道写端。代码中是这样处理的：

if (!channels.empty()) {
    for (auto& ch : channels) {
        ch.ClosePipe();
    }
}

最终的效果是：每个管道只对应一个子进程，形成 1:1 的专属通信关系。

八、任务执行流程

Dotask()

void Dotask(int fd) {
    while(true) {
        read(fd, &task_code, sizeof(task_code));
        // 校验后执行 tasks[task_code]();
    }
}

子进程阻塞在 read 上，直到父进程往对应的管道写端写入了任务码。子进程读到任务码后，比如读到 2，就会去调用 tasks[2]()，也就是 PrintMessage()。

整个链路很清晰：父进程 write(2) → 管道 → 子进程 read() → tasks[2]() → PrintMessage()。

九、任务调度策略

这里的调度策略相对简单粗暴，但足以说明问题：

• SelectTask()：通过 rand() % 4 随机选择一个任务码（0~3）。
• SelectChannel()：采用经典的 Round Robin（轮询） 策略。static int index 从0开始，每次取模 channels 的大小，依次选中每个子进程。

执行顺序就是：0, 1, 2, 3, 0, 1, 2, 3... 保证所有子进程均衡地获得任务。

十、运行流程图

┌───────────┐
│  Run()    │
└─────┬─────┘
      │
      ▼
选择任务 (SelectTask)
      │
      ▼
选择Channel (SelectChannel)
      │
      ▼
Write(任务码) 到对应管道
      │
      ▼
子进程read() 阻塞返回
      │
      ▼
执行任务函数
      │
      ▼
继续等待下一个任务

十一、退出流程

优雅退出是进程池设计里不可忽视的一环。流程是这样的：

父进程：for (auto& ch : channels) { ch.ClosePipe(); }
子进程：read() 返回 0（EOF），break 退出循环。
父进程：for (auto& ch : channels) { ch.Wait(); }  // waitpid 回收资源

父进程先关闭所有管道写端，一旦写端关闭，子进程的 read 就会返回0，表示“没有任务了”，子进程据此退出循环。然后父进程调用 waitpid 回收子进程资源。完成。

十二、源码中的亮点

这个实现里有一个地方值得单独拎出来说——回调函数的设计。它用的是 std::function 作为回调类型：

using cb_t = std::function;
pool.Init(Dotask);
// 在CreateProcessChannel中，子进程调用 cb(pipefd[0])

这样做的好处是什么？进程池的框架和具体的业务逻辑实现了解耦。框架负责进程创建、管道管理、任务分发和资源回收；而业务逻辑，也就是子进程拿到任务后具体要做什么，完全由传入的回调函数 Dotask 决定。想换一套任务处理逻辑？换一个函数指针就行了。