Linux 基于匿名管道实现简易进程池

Linux 匿名管道内存上限测试显示系统默认约为 64KB。文章详解利用匿名管道配合 fork 创建进程池的方法，通过父进程向子进程发送任务码实现单向通信。核心类 Channel 封装读写端，ChannelManager 维护进程列表并轮询分发任务，ProcessPool 负责生命周期管理。该方案预分配进程资源，减少频繁创建销毁开销，提升执行效率。

咸鱼开飞机发布于 2026/1/17更新于 2026/4/211 浏览

1. 管道内存上限

管道并非无限内存，而是有容量限制的。通过测试可以确定具体大小。

#include <iostream>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <cstdio>
#include <cstring>

void ChildWrite(int wfd) {
    char c = 0;
    int cnt = 0;
    while (true) {
        write(wfd, &c, 1);
        std::cout << "cnt : " << cnt++ << std::endl;
    }
}

int main() {
    int pfd[2] = {0};
    if (pipe(pfd) == -1) {
        std::cerr << "pipe error!" << std::endl;
        return 1;
    }

    pid_t id = fork();
    if (id == 0) {
        close(pfd[0]);
        ChildWrite(pfd[1]);
        (pfd[]);
    }  {
        (pfd[]);
        (id, , );
        (pfd[]);
    }
     ;
}

相关免费在线工具

Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。在线工具，Base64 字符串编码/解码在线工具，online
Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。在线工具，Base64 文件转换器在线工具，online
Markdown转HTML
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。在线工具，Markdown转HTML在线工具，online
HTML转Markdown
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。在线工具，HTML转Markdown在线工具，online
JSON 压缩
通过删除不必要的空白来缩小和压缩JSON。在线工具，JSON 压缩在线工具，online
JSON美化和格式化
将JSON字符串修饰为友好的可读格式。在线工具，JSON美化和格式化在线工具，online

操作	目的 & 原理
父进程创建多个管道	为每个子进程分配专属单向通信通道，用于发送任务码
fork 多个子进程	预创建一批子进程，避免频繁创建销毁开销
父关读端、子关写端	确保管道是「父写、子读」的单向通信，符合半双工特性
父写任务码→子执行	子进程阻塞在 `read()` 上等待任务，收到后执行并继续阻塞

processpool : Main.cc
	g++ $^ -o $@ -std=c++11
.PHONY : clean
clean :
	rm -f processpool

#include <iostream>
#include <string>
#include <cstdlib>
#include <vector>
#include <unistd.h>
#include <sys/wait.h>
#include "Task.hpp"

class Channel {
private:
    int _wfd;
    int _id;
    std::string _name;
public:
    Channel(int wfd, int id) : _wfd(wfd), _id(id) {
        _name = "channel -- " + std::to_string(_wfd) + " -- " + std::to_string(_id);
    }
    ~Channel() {}
    std::string Name() { return _name; }
    void Send(int taskcode) {
        int n = write(_wfd, &taskcode, sizeof(taskcode));
        (void)n;
    }
    void CloseFatherProcess() {
        std::cout << "父进程写端 : " << _wfd << " 关闭" << std::endl;
        close(_wfd);
    }
    void WaitChildProcess() {
        pid_t rid = waitpid(_id, nullptr, 0);
        (void)rid;
    }
};

class ChannelManager {
private:
    std::vector<Channel> _channels;
    int _next;
public:
    ChannelManager() : _next(0) {}
    ~ChannelManager() {}
    void Insert(int wfd, pid_t id) {
        _channels.push_back({wfd, id});
    }
    void Print() {
        for (auto &e : _channels) {
            std::cout << e.Name() << std::endl;
        }
    }
    Channel& Select() {
        auto& e = _channels[_next];
        _next++;
        _next %= _channels.size();
        return e;
    }
    void CloseSubFatherProcess() {
        for (auto& e : _channels) {
            e.CloseFatherProcess();
        }
    }
    void WaitSubChildProcess() {
        for (auto& e : _channels) {
            e.WaitChildProcess();
        }
    }
};

class ProcessPool {
public:
    ProcessPool(int num = 5) : _process_num(num) {
        _tm.Register(PrintLog);
        _tm.Register(DownLoad);
        _tm.Register(UpLoad);
    }
    ~ProcessPool() {}
    void Work(int rfd) {
        while (true) {
            int taskcode = 0;
            ssize_t n = read(rfd, &taskcode, sizeof(taskcode));
            if (n > 0) {
                if (n != sizeof(taskcode)) continue;
                std::cout << "子进程 [" << getpid() << "] 收到一个任务码 : " << taskcode << std::endl;
                _tm.Execute(taskcode);
            } else if (n == 0) {
                std::cout << "子进程 [" << getpid() << "] 退出" << std::endl;
                break;
            } else {
                std::cout << "读取错误" << std::endl;
                break;
            }
        }
    }
    bool Start() {
        for (int i = 0; i < _process_num; i++) {
            int pipefd[2] = {0};
            if (pipe(pipefd) < 0) return false;
            pid_t id = fork();
            if (id < 0) {
                return false;
            } else if (id == 0) {
                close(pipefd[1]);
                Work(pipefd[0]);
                exit(0);
            } else {
                close(pipefd[0]);
                _cm.Insert(pipefd[1], id);
            }
        }
        return true;
    }
    void DeBug() { _cm.Print(); }
    void Run() {
        int taskcode = _tm.Code();
        auto& e = _cm.Select();
        std::cout << "选择了一个子进程 : " << e.Name() << std::endl;
        std::cout << "发射了一个任务码 : " << taskcode << std::endl;
        e.Send(taskcode);
    }
    void Stop() {
        _cm.CloseSubFatherProcess();
        _cm.WaitSubChildProcess();
    }
private:
    ChannelManager _cm;
    TaskManager _tm;
    int _process_num;
};

#include "ProcessPool.hpp"

int main() {
    ProcessPool p(5);
    p.Start();
    int cnt = 10;
    while (cnt--) {
        p.Run();
        sleep(1);
        std::cout << "---------------------------" << std::endl;
    }
    p.Stop();
    return 0;
}

#include <iostream>
#include <vector>
#include <ctime>

void PrintLog() {
    std::cout << "我是一个打印日志的任务" << std::endl;
}
void DownLoad() {
    std::cout << "我是一个下载的任务" << std::endl;
}
void UpLoad() {
    std::cout << "我是一个上传的任务" << std::endl;
}

typedef void (*task_t)();

class TaskManager {
public:
    TaskManager() { srand(time(nullptr)); }
    void Register(task_t t) { _tasks.push_back(t); }
    int Code() { return rand() % _tasks.size(); }
    void Execute(int taskcode) {
        if (taskcode >= 0 && taskcode < _tasks.size()) {
            _tasks[taskcode]();
        }
    }
    ~TaskManager() {}
private:
    std::vector<task_t> _tasks;
};

Linux 基于匿名管道实现简易进程池

1. 管道内存上限

更多推荐文章

相关免费在线工具

2. 进程池

1. 什么是进程池

2. 操作原理

3. 实现代码

Linux 基于匿名管道实现简易进程池

1. 管道内存上限

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

2. 进程池

1. 什么是进程池

2. 操作原理

3. 实现代码