【Linux笔记】进程间通信——匿名管道||进程池

2025-04-28 04:37:19 0 阅读

🔥个人主页🔥：孤寂大仙V
🌈收录专栏🌈：Linux
🌹往期回顾🌹：【Linux笔记】动态库与静态库的理解与加载
🔖流水不争，争的是滔滔不

一、Linux进程间通信简介和目的
二、常见IPC方式
- 管道
- System V
三、匿名管道
- 工作原理
- 特性
- 管道读写规则
四、基于匿名管道的进程池

一、Linux进程间通信简介和目的

简介

进程间通信（IPC，Inter-Process Communication）是操作系统中不同进程之间交换数据、协调工作的核心机制。由于每个进程拥有独立的内存空间，彼此隔离，因此需要特定的技术手段实现跨进程协作。

进程间通信(IPC)是指运行在同一台计算机或不同计算机上的多个进程之间进行数据交换和通信的技术。由于每个进程都有自己的地址空间，它们无法直接访问彼此的数据，因此需要通过特定的机制实现通信。IPC是操作系统和多进程编程中的一个重要概念，广泛应用于分布式系统、多任务操作系统以及各种应用程序之间。

进程间通信的本质：是先让不同的进程，先看到同一份资源，然后才有通信的条件。

目的

数据传输：一个进程需要将它的数据发送给另一个进程
资源共享：多个进程之间共享同样的资源。
通知事件：一个进程需要向另一个或一组进程发送消息，通知它（它们）发生了某种事件（如进程终止时要通知父进程）。
进程控制：有些进程希望完全控制另一个进程的执行（如Debug进程），此时控制进程希望能够拦截另一个进程的所有陷入和异常，并能够及时知道它的状态改变

二、常见IPC方式

管道

类型：

匿名管道：用于父子进程间通信（半双工）。
有名管道（FIFO）：可用于不相关进程（全双工）。

特点：

内核缓冲区存储数据，读写同步。
数据读取后即删除，不支持随机访问。

System V

共享内存（Shared Memory）
核心机制：多个进程直接访问同一块物理内存。
特点：
速度最快，无需数据复制。
需配合同步机制（如信号量、互斥锁）。
消息队列（Message Queue）
机制：进程通过发送 / 接收消息（带类型）通信。
特点：
消息按类型排序，支持异步通信。
消息队列在内核中持久化，直到被显式删除。
信号量（Semaphore）
作用：控制对临界资源的访问，实现同步与互斥。
类型：
二元信号量：0/1 状态（互斥锁）。
计数信号量：控制并发访问数量。

三、匿名管道

我们把从一个进程连接到另一个进程的一个数据流称为一个“管道”

工作原理

通过父子进程继承关系，再将文件描述符关闭，实现一端写，一端读，就是匿名管道。

匿名管道是单向通信（半双工）数据只能从管道的一端写入，另一端读取，无法双向传输。内核缓冲区存储数据通过内核临时存储，大小通常为 64KB（可通过 ulimit -a 查看）。生命周期短，管道随创建它的进程（父进程）结束而自动销毁，无需手动清理。亲缘关系限制仅适用于父子进程或兄弟进程（通过 fork/exec 系列函数创建的进程）。

父进程以读写方式打开文件。父进程fork创建子进程，（进程具有独立性）子进程要拷贝一份PCB结构，PCB中包含了files_struct结构,files_struct中有一个指向struct file（文件）的指针数组，而文件描述符就是这个数组的下标。在拷贝后，子进程也就有了指向struct file（文件）的对应数组元素下标（文件描述符）。而struct file（文件）是独属于文件的，和进程没有关系，也就不用拷贝，也就是说此时父子进程公共区域就是 struct file。（实现不同进程看到同一份资源）。write是系统调用接口，会将数据放在内核缓冲区，底层定期刷新缓冲区将内容写到磁盘。

创建管道
通过系统调用 pipe(int fd[2]) 创建匿名管道，返回两个文件描述符：
fd[0]：读端（read end）。
fd[1]：写端（write end）。

数据流动
父进程通过 write(fd[1], buf, len) 向管道写入数据。
子进程通过 read(fd[0], buf, len) 读取数据。

子进程继承父进程的文件描述符表

匿名管道的示例

#include 
#include 
#include 
#include 
#include 
#include 

// 9:
void ChildWrite(int wfd)//子进程写
{
    char c = 0;
    int cnt = 0;
    char buffer[1024];
    while (true)
    {
        
        snprintf(buffer, sizeof(buffer), "I am child, pid: %d, cnt: %d
", getpid(), cnt++);
        write(wfd,buffer,strlen(buffer));
    }
}

void FatherRead(int rfd)//父进程读
{
    char buffer[1024];
    while (true)
    {
        // sleep(100);
        sleep(1);
        buffer[0] = 0;
        ssize_t n = read(rfd, buffer, sizeof(buffer)-1);
        if(n > 0)
        {
            buffer[n] = 0;
            std::cout << "child say: " << buffer << std::endl;
            // sleep(2);
        }
        else if(n == 0)
        {
            std::cout << "n : " << n << std::endl;
            std::cout << "child 退出，我也退出";
            break;
        }
        else
        {
            break;
        }

        break;
    }
}

int main()
{
    // 1. 创建管道
    int fds[2] = {0}; // fds[0]:读端   fds[1]: 写端
    int n = pipe(fds);
    if (n < 0)
    {
        std::cerr << "pipe error" << std::endl;
        return 1;
    }
    std::cout << "fds[0]: " << fds[0] << std::endl;
    std::cout << "fds[1]: " << fds[1] << std::endl;

    // 2. 创建子进程
    pid_t id = fork();
    if (id == 0)
    {
        // child
        // code
        // 3. 关闭不需要的读写端，形成通信信道
        // f -> r, c -> w
        close(fds[0]);
        ChildWrite(fds[1]);
        close(fds[1]);
        exit(0);
    }
    // 3. 关闭不需要的读写端，形成通信信道
    // f -> r, c -> w
    close(fds[1]);
    FatherRead(fds[0]);
    close(fds[0]);


    int status = 0;
    int ret = waitpid(id, &status, 0); // 获取到子进程的退出信息吗！！！
    if(ret > 0)
    {
        printf("exit code: %d, exit signal: %d
", (status>>8)&0xFF, status&0x7F);
    }
    return 0;
}

以上代码通过子进程写父进程写，形成信道。

特性

匿名管道，只能用来进行具有血缘关系的进程进行进程间通信。
管道是面向字节流的
管道是单向通信的-需要半双工的一种特殊情况
管道文件自带同步机制

同步机制的四种情况：

写入慢，读取快。读端就要阻塞，等写端的操作。

读端等待写端进行操作，写一条读一条。

写入快，读取慢。读端满了，写端就要阻塞。

把读取端也就是父进程，设置为slee(5),一瞬间写端子进程写满了，等待读端进程进行读取

写关，读端继续读。read就会读到返回值为0，表示文件的结尾。
读端关闭，写端继续写。写端的写入无意义。操作系统会杀掉进程。

管道读写规则

当没有数据可读时

O_NONBLOCK disable：read调用阻塞，即进程暂停执行，一直等到有数据来到为止。
O_NONBLOCK enable：read调用返回-1，errno值为EAGAIN。

当管道满的时候

O_NONBLOCK disable： write调用阻塞，直到有进程读走数据
O_NONBLOCK enable：调用返回-1，errno值为EAGAIN
如果所有管道写端对应的文件描述符被关闭，则read返回0
如果所有管道读端对应的文件描述符被关闭，则write操作会产生信号SIGPIPE,进而可能导致write进程

退出

当要写入的数据量不大于PIPE_BUF时，linux将保证写入的原子性。
当要写入的数据量大于PIPE_BUF时，linux将不再保证写入的原子性。

四、基于匿名管道的进程池

池化技术（Pooling Technology）是一种通过预先创建并复用资源实例来优化系统性能的技术。其核心思想是减少资源频繁创建与销毁的开销，提升资源利用率。以下是关于池化技术的详细说明：

预创建资源：在系统初始化时，预先创建一定数量的资源实例（如进程、线程、数据库连接等）。
复用机制：当任务需要资源时，直接从池中获取已创建的实例，而非新建；任务完成后，资源归还池内供后续任务使用。
动态管理：部分池化技术支持根据负载动态调整资源数量（如扩容或缩容）。

进程池（Process Pool）适用场景：CPU 密集型任务（如并行计算、批量数据处理）。
特点：每个进程独立拥有内存空间，避免全局变量冲突。适合多核 CPU，但内存占用较高。

以下是基于匿名管道实现的简单进程池

#ifndef _PROCESS_POOL_HPP_
#define _PROCESS_POOL_HPP_

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "Task.hpp"

using namespace std;

const int Poolnum = 5;
class Channel
{
public:
    Channel(int fd, pid_t pid)
        : _wfd(fd), _supid(pid)

    ~Channel() {}

private:

};

class ChannelManager
{
public:
    ChannelManager()
        : _next()
    {
    }

    ~ChannelManager() {}

private:
}

class ProcessPool
{
public:
    ProcessPool()
        : _process_num()
    {
    }
    void work(int rfd)
    {

    }
    bool Start()
    {
        for (int i = 0; i < _process_num; i++)
        {
            int pipefd[2] = {0}; // 创建管道
            int n = pipe(pipefd);
            if (n < 0)
            {
                return false;
            }
            pid_t supid = fork(); // 创建进程
            if (supid < 0)
            {
                return false;
            }
            else if (supid == 0)
            {
                // 子进程读
                close(pipefd[1]); // 关闭子进程写端
                work(pipefd[0]);
                close(pipefd[0]);
                exit(1);
            }
            else
            {
                // 父进程写
                close(pipefd[0]); // 关闭子进程读端
                _cm.Insert(pipefd[1], supid);
            }
        }
        return true;
    }

    ~ProcessPool() {}

private:
};
#endif

整体框架是，有一个channel类是对单个的信道进行描述的，一个channelmanager类是对多个信道进行描述，最后是processpool是对进程的读写以及整体进行描述的。

以下是procress这个类内部函数的理解

bool Start()
    {
        for (int i = 0; i < _process_num; i++)
        {
            int pipefd[2] = {0}; // 创建管道
            int n = pipe(pipefd);
            if (n < 0)
            {
                return false;
            }
            pid_t supid = fork(); // 创建进程
            if (supid < 0)
            {
                return false;
            }
            else if (supid == 0)
            {
                // 子进程读
                close(pipefd[1]); // 关闭子进程写端
                work(pipefd[0]);
                close(pipefd[0]);
                exit(1);
            }
            else
            {
                // 父进程写
                close(pipefd[0]); // 关闭子进程读端
                _cm.Insert(pipefd[1], supid);
            }
        }
        return true;
    }

这里创建进程和匿名管道，让子进程读父进程写。注意信道是多个的数量由你自己来设定，所以设置一个for循环进行创建多个子进程与信道。把父进程的写端和进程的pid传给具体的channel信道，具体的工作让具体的channel信道完成。

 ProcessPool(int num)
        : _process_num(num)
    {

    }
    void work(int rfd)
    {
        int code = 0;

        ssize_t n = read(rfd, &code, sizeof(code));
        if (n > 0)
        {
            if (n != sizeof(code))
            {
                return;
            }
            cout << "子进程pid->" << getpid() << "收到任务码->" << code << endl;
            _tm.Execute(code);
        }
        else if (n == 0)
        {
            cout << "子进程退出" << endl;
        }
        else
        {
            cout << "读取错误" << endl;
        }
    }

上面是构造和子进程的读端是如何工作的，在这里设置一个任务码code，让子进程根据任务码的不同去做不同的工作。

void Run()
    {
        int taskcode = _tm.Code(); // 选择一个任务
        auto &c = _cm.Select();    // 选择一个信道
        c.Send(taskcode);          // 发送任务码
    }
    void Stop()
    {
        // 父进程读端关闭
        _cm.StopSubProcess();
        // 回收子进程
        _cm.WaitSubProcess();
    }

上面代码是选择任务码，子进程需要收到任务码然后去做不同的工作，发给信道然后对应的子进程接收任务吗完成工作。这里在任务函数中在详细聊。执行任务完成后父进程读端和子进程写端都需要关闭。

下面是channel类

class Channel
{
public:
    Channel(int fd, pid_t pid)
        : _wfd(fd), _supid(pid)
    {
        _name = "Channel-" + to_string(_wfd) + "-" + to_string(_supid);
    }

    void Send(int code)
    {
        int n = write(_wfd, &code, sizeof(code));
        (void)n;
    }
    void Close()
    {
        close(_wfd);
    }
    void Wait()
    {
        pid_t rid = waitpid(_supid, nullptr, 0);
        (void)rid;
    }
    string Name()
    {
        return _name;
    }

    ~Channel() {}

private:
    int _wfd;
    pid_t _supid;
    string _name;
};

构造接收写端和pid，send()函数对应processpool类中的发送任f务码，要在具体的channel信道中发送任务码给子进程，每个子进程才能接受到对应的任务码。这里的close和wait就是关闭写端和回收子进程，对应processpool中的void Stop()。

下面是ChannelManager类

class ChannelManager
{
public:
    ChannelManager(int next = 0)
        : _next(next)
    {
    }
    void Insert(int fd, int pid)
    {
        _channel.emplace_back(fd, pid);
    }

    Channel &Select()
    {
        auto &c = _channel[_next];
        _next++;
        _next %= _channel.size();
        return c;
    }

    void StopSubProcess()
    {
        for (auto &channel : _channel)
        {
            channel.Close();
            std::cout << "关闭: " << channel.Name() << std::endl;
        }
    }
    void WaitSubProcess()
    {
        for (auto &channel : _channel)
        {
            channel.Wait();
            std::cout << "回收: " << channel.Name() << std::endl;
        }
    }
    ~ChannelManager() {}

private:
    vector<Channel> _channel;
    int _next;
};

对通信信道用vector数组进程管理。 Channel &Select()，是选择任务码的时候在整个通信信道的数组中采用沦陷的方式进行选择信道。 void Insert(int fd, int pid)就是传过来写端和进程pid与channel的构造相关联与processpool的父进程传写端和pid相关联。void StopSubProcess() void WaitSubProcess()就是关闭写端和等待子进程与close()和wait()相关联。

任务码具体实现的函数

#pragma once
#include
#include
using namespace std;

typedef void(*task_t)();//函数指针

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


class TaskManager
{
public:
    TaskManager()
    {
        srand(time(nullptr)); //时间戳种子
    }

    int Code()//产生随机值
    {
        return rand()%_tasks.size();
    }

    void Register(task_t t)//注册任务
    {
        _tasks.push_back(t);
    }

    void Execute(int code)
    {
        if(code>=0 && code<_tasks.size())
        {
            _tasks[code]();
        }
    }
private:
    vector<task_t> _tasks;
};

用函数指针，根据时间戳产生随机值，然后把对应注册的任务放进存储任务的数组中，最后execute函数根据产生的任务码在储存任务的数组中选择任务。

子进程work就是调用execute()把收到的任务码给这个函数然后让这个函数进行选择对应的任务。

   void Run()
    {
        int taskcode = _tm.Code(); // 选择一个任务
        auto &c = _cm.Select();    // 选择一个信道
        c.Send(taskcode);          // 发送任务码
    }

在次重新聊一下processpool中的run函数，通过调用task.hpp的code函数，获得任务码，然后通过Select()函数选择以一个信道，然后发送任务码给子进程，子进程接受任务码然后子进程开始运行。

ProcessPool.hpp

#ifndef _PROCESS_POOL_HPP_
#define _PROCESS_POOL_HPP_

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "Task.hpp"
using namespace std;

const int Poolnum = 5;
class Channel
{
public:
    Channel(int fd, pid_t pid)
        : _wfd(fd), _supid(pid)
    {
        _name = "Channel-" + to_string(_wfd) + "-" + to_string(_supid);
    }

    void Send(int code)
    {
        int n = write(_wfd, &code, sizeof(code));
        (void)n;
    }
    void Close()
    {
        close(_wfd);
    }
    void Wait()
    {
        pid_t rid = waitpid(_supid, nullptr, 0);
        (void)rid;
    }
    string Name()
    {
        return _name;
    }

    ~Channel() {}

private:
    int _wfd;
    pid_t _supid;
    string _name;
};

class ChannelManager
{
public:
    ChannelManager(int next = 0)
        : _next(next)
    {
    }
    void Insert(int fd, int pid)
    {
        _channel.emplace_back(fd, pid);
    }

    Channel &Select()
    {
        auto &c = _channel[_next];
        _next++;
        _next %= _channel.size();
        return c;
    }

    void StopSubProcess()
    {
        for (auto &channel : _channel)
        {
            channel.Close();
            std::cout << "关闭: " << channel.Name() << std::endl;
        }
    }
    void WaitSubProcess()
    {
        for (auto &channel : _channel)
        {
            channel.Wait();
            std::cout << "回收: " << channel.Name() << std::endl;
        }
    }
    ~ChannelManager() {}

private:
    vector<Channel> _channel;
    int _next;
};

class ProcessPool
{
public:
    ProcessPool(int num)
        : _process_num(num)
    {
        _tm.Register(PrintLog);
        _tm.Register(DownLoad);
        _tm.Register(UpLoad);
    }
    void work(int rfd)
    {
        int code = 0;

        ssize_t n = read(rfd, &code, sizeof(code));
        if (n > 0)
        {
            if (n != sizeof(code))
            {
                return;
            }
            cout << "子进程pid->" << getpid() << "收到任务码->" << code << endl;
            _tm.Execute(code);
        }
        else if (n == 0)
        {
            cout << "子进程退出" << endl;
        }
        else
        {
            cout << "读取错误" << endl;
        }
    }
    bool Start()
    {
        for (int i = 0; i < _process_num; i++)
        {
            int pipefd[2] = {0}; // 创建管道
            int n = pipe(pipefd);
            if (n < 0)
            {
                return false;
            }
            pid_t supid = fork(); // 创建进程
            if (supid < 0)
            {
                return false;
            }
            else if (supid == 0)
            {
                // 子进程读
                close(pipefd[1]); // 关闭子进程写端
                work(pipefd[0]);
                close(pipefd[0]);
                exit(1);
            }
            else
            {
                // 父进程写
                close(pipefd[0]); // 关闭子进程读端
                _cm.Insert(pipefd[1], supid);
            }
        }
        return true;
    }
    void Run()
    {
        int taskcode = _tm.Code(); // 选择一个任务
        auto &c = _cm.Select();    // 选择一个信道
        c.Send(taskcode);          // 发送任务码
    }
    void Stop()
    {
        // 父进程读端关闭
        _cm.StopSubProcess();
        // 回收子进程
        _cm.WaitSubProcess();
    }
    ~ProcessPool() {}

private:
    int _process_num = 5;
    ChannelManager _cm;
    TaskManager _tm;
};
#endif

Task.hpp

#pragma once
#include
#include
using namespace std;

typedef void(*task_t)();//函数指针

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


class TaskManager
{
public:
    TaskManager()
    {
        srand(time(nullptr)); //时间戳种子
    }

    int Code()//产生随机值
    {
        return rand()%_tasks.size();
    }

    void Register(task_t t)//注册任务
    {
        _tasks.push_back(t);
    }

    void Execute(int code)
    {
        if(code>=0 && code<_tasks.size())
        {
            _tasks[code]();
        }
    }
private:
    vector<task_t> _tasks;
};

main.cc

#include "ProcessPool.hpp"
int main()
{
    ProcessPool pp(Poolnum);
    pp.Start();
    int ch=5;
    while(ch--)
    {
        pp.Run();
        sleep(2);
    }
    pp.Stop();
}