Linux：进程池制作（匿名管道版本 & 定名管道版本）

每次来任务才 fork 一个进程，创建成本会直接摊到业务延迟上。进程池的思路正相反：启动前先批量造好一批工作进程，任务来了直接分发给空闲进程执行。这篇用匿名管道实现一个进程池，并顺带讲清管道管理、fd 泄漏和子进程退出这三个最容易翻车的地方。

一、为什么要进程池

创建进程是要花时间的，fork 一次内核要做进程描述符、页表、缓冲区等一系列复制动作（现代内核用写时拷贝把成本压得很低，但反复创建销毁依然可观）。如果任务本身只有毫秒级，创建进程的开销就会盖过任务本身。

进程池的做法是：进程启动时先初始化一批子进程，平时它们阻塞在管道读端等着，来任务时父进程把「任务码」写进某个子进程对应的管道，子进程读到就执行，执行完继续等下一次。这些提前创建好的进程集合就是进程池。

二、整体框架：先描述，再组织

父进程要同时知道：往哪根管道写、对应哪个子进程、这个子进程叫什么。所以第一步是给「一根管道 + 一个 worker」做个结构体，再用 vector 装起来，便于快速随机访问。

#define NUM 5          // 任务进程个数

int number = 1;         // 管道编号

class channel
{
public:
    channel(int fd, pid_t id)
        : ctrlfd(fd)
        , workerid(id)
    {
        name = "channel-" + std::to_string(number++);
    }

public:
    int ctrlfd;
    pid_t workerid;
    std::string name;
};

主流程就三步：

int main()
{
    std::vector<channel> channels;   // 用 vector 管理管道

    // 1、创建管道，创建进程，指定工作进程的工作方式
    CreaterChannel(&channels);

    // 2、主进程向工作进程发送任务
    const bool g_always_loop = true;
    SendTask(channels, !g_always_loop, 10);

    // 3、回收资源：关闭写端，子进程自动退出
    ReleaseChannels(channels);
    return 0;
}

三、创建管道与子进程（匿名管道版）

父进程循环 NUM 次，每次先建管道再建子进程，然后关掉双方不需要的读写端，形成一条单向通道。核心代码：

void CreaterChannel(std::vector<channel>* channels)
{
    std::vector<int> old;      // 记录子进程继承的多余读端
    for (int i = 0; i < NUM; i++)   // 创建 NUM 个工作进程
    {
        // 1. 创建管道
        int pipefd[2];
        int n = pipe(pipefd);
        assert(n == 0);
        (void)n;    // 防止编译器报警

        // 2. 创建子进程
        pid_t id = fork();
        if (id < 0)
        {
            perror("fork");
            return;
        }

        // 3. 形成单向通信
        if (id == 0)     // 子进程，也就是工作进程
        {
            if (!old.empty())    // 将子进程继承父进程的多余读端全部关闭
            {
                for (auto rfd : old)
                {
                    close(rfd);
                }
            }
            close(pipefd[1]);
            dup2(pipefd[0], 0);  // 把读端重定向到标准输入
            work();              // 子进程执行任务
            exit(0);
        }

        // 父进程
        close(pipefd[0]);
        channels->push_back(channel(pipefd[1], id));  // 记录写端与 worker 信息
        old.push_back(pipefd[1]);
    }
}

3.1 那个必须处理的 fd 泄漏

简单「关掉不需要的读写端」是不够的，因为 fork 出来的子进程会继承父进程当时打开的所有 fd。后面创建的子进程又继承了前面子进程的 fd，一层层传下去，最后每个子进程手里都攥着一堆并行的管道读端，导致管道一直无法彻底关闭，资源泄漏。

解决办法是：父进程每建一个写端就记进 old，新子进程一诞生，就把它从父进程继承来的多余读端全部 close 掉。这一步是进程池能不能长期稳定运行的关键。

四、工作进程怎么干活

子进程循环从标准输入（已经被 dup2 到管道读端）读 4 字节的任务码：读到就校验任务码、执行任务；读返回 0 说明父进程写端全关了，管道写端闭合，read 返回 0，子进程该退出了。

void work()
{
    while (1)
    {
        int code = 0;
        ssize_t n = read(0, &code, sizeof(code));
        if (n == 0)
        {
            // 主进程写端退出，子进程也退出
            break;
        }
        else if (n > 0)
        {
            if (!init.CheckSafe(code))   // 检查任务码是否合法
                continue;
            init.RunTask(code);          // 执行任务
        }
        else
        {
            // nothing
        }
    }
}

约定父进程一次只写 4 字节、子进程一次只读 4 字节，任务码就是一个 int，这样天然不会粘包，也不用关心半包。

五、发任务与资源回收

父进程遍历 channels，按策略（轮询或随机）挑一个 worker，往它的 ctrlfd 写一个任务码即可。用一个开关变量控制是「一直发」还是「发够 N 次就停」：

SendTask(channels, !g_always_loop, 10);

收尾时把父进程这边的写端逐个关掉。写端全部关闭后，管道读端会读到 0，上面 work() 里的循环自然退出，子进程跟着退出，父进程再 wait 回收，整套进程池干净退场。

六、换成命名管道版怎么改

• 管道从匿名（pipe()，靠 fd 继承）换成命名管道（mkfifo()，靠路径），父子进程不需要有亲缘关系也能通信。
• 描述符管理从「继承后逐个 close」变成「按路径打开、用完按路径关闭」，泄漏风险小很多。
• 结构体里 ctrlfd 的来源改为 open(fifoPath, O_WRONLY)，其余框架、任务码校验、work() 循环完全不用动。

所以两种版本的差异只在「通道怎么建」，职责划分（描述 channel、组织 vector、派发任务、回收资源）一模一样。

七、踩坑清单

• 不关继承的多余读端 → 子进程永不退出、fd 与管道资源泄漏，长时间运行必然出问题。
• 忘了 dup2 到 0，work() 里写死 read(0, ...) 就会读到终端输入而不是管道。
• 任务码定长但用了字符串，长度不固定就得自己做粘包处理，直接用 int 最省事。
• 父进程主流程退出前必须先关写端，否则子进程会一直阻塞在 read 上变成僵尸。
• 进程池大小别盲目开大：worker 太多只是增加上下文切换开销，一般和 CPU 核数、任务类型（CPU 密集 / IO 密集）挂钩来定。

qidao123.com企服评测-应用市场 编辑提示