目录

1. 主函数

2. 要执行的任务

3. 创建信道

4. 发送消息

5. 回收管道

---

创建和销毁进程需要使用系统调用，操作系统分配资源（内存、文件描述符等），开销较大。我们可以提前创建好一些进程，当操作系统需要完成一些任务时可以直接分配，不用再单独一个个创建，减小一部分开销。

内存池：申请的空间往往是大于你要求的，可能你暂时用不到，想用的时候就可以直接用，不用再申请空间。但是这些空间是需要交给用户管理的，我们使用 channel 描述管理进行管道：

class channel//先描述再组织
{    
public:
    channel(int fd,pid_t id):ctrlfd(fd),workerid(id)
    {
        name = "channel-"+std::to_string(number);
    }

    int ctrlfd;//父进程中对应的文件描述符
    pid_t workerid;//子进程id
    std::string name;//管道名
};

1. 主函数

我们创建一个信道集合 channels ，方便统一管理。然后创建信道，父进程发送任务，子进程执行，之后父进程关闭写端回收子进程，回收资源。

int main()
{
    std::vector<channel> channels;
    //1.创建信道，创建进程
    CreateChannels(&channels);
    
    //2.父进程，发送任务
    const bool g_always_loop = true;
    //SendCommand(channels,g_always_loop);//一直运行
    SendCommand(channels,!g_always_loop,10);//发送10次任务

    //3.回收资源
    ReleaseChannels(channels);

    return 0;
}

2. 要执行的任务

这里模拟定义了4种任务，我们调用采用随机执行的方法。

#pragma once

#include <iostream>
#include <functional>
#include <vector>
#include <unistd.h>

//using task_t std::function<void>;
typedef std::function<void()> task_t;

void Download()
{
    std::cout<<"我是一个下载任务,处理者：" << getpid() << std::endl;
}
void PrintLog()
{
    std::cout<<"我是一个打印日志任务,处理者：" << getpid() << std::endl;
}
void PushVideoStream()
{
    std::cout<<"我是一个推送视频任务,处理者：" << getpid() << std::endl;
}

class Init
{
public:
    // 任务集合
    std::vector<task_t> tasks; //成员函数可以访问静态成员，静态不能访问非静态
    // 任务集合 -- 没用到
    const static int g_download_code = 0;
    const static int g_printLog_code = 1;
    const static int g_push_video_stream_code = 2;
public:
    Init()
    {
        tasks.push_back(Download);
        tasks.push_back(PrintLog);
        tasks.push_back(PushVideoStream);
        srand(time(nullptr)^getpid());//更随机
    }

    bool CheckSafe(int code)
    {
        if(code >= 0 && code < 3)
            return true;
        else
            return false;

    }
    void RunTask(int code)
    {
        tasks[code]();
    }
    int SelectTask()//随机选择任务
    {
        return rand()%tasks.size();
    }
    std::string ToDes(int code)
    {
        switch (code)
        {
        case g_download_code:
            return "download";
            break;
        case g_printLog_code:
            return "PrintLog";
            break;
        case g_push_video_stream_code:
            return "push_video_stream";
            break;
        default:
            return " ";
            break;
        }
    }
};
//静态成员类外定义，或者静态函数中push_back
//std::vector<task_t> Init::tasks = {Download, PrintLog, PushVideoStream};

Init init;//定义对象

3. 创建信道

我们这里创建有 5 个子进程的进程池。temp 用来存储父进程打开的所有子进程写端，它的作用可以让每个子进程关闭父进程之前创建的其他子进程读端 fd ，即与自己无关的文件描述符，为什么我们在后面回收管道中讲。

然后对于每个子进程我们将标准输入重定向到读端文件，以后子进程读端读数据只需要从 fd = 0 中 read 读去就行。

最后对每个管道都添加到 channels 进程池中，方便管理。

const int num = 5;
static int number = 1;

//创建信道，创建进程
void CreateChannels(std::vector<channel>* c)
{
    //bug
    std::vector<int> temp;
    for(int i = 0;i<num;i++)
    {
        //1.先创建管道
        int pipefd[2];
        int n = pipe(pipefd);//不能传入pipefd[2],pipefd[2]是一个pipefd数组的第3个值，已经越界.
        assert(n == 0);

        //2.创建进程
        pid_t id = fork();
        assert(id != -1);

        //3.构建单向通信的信道
        //3.1 子进程    子读父写
        if(id == 0)
        {   
            if(!temp.empty())//每个子进程关闭 父进程之前创建的其他子进程读端fd 即与自己无关的文件描述符
            {
                for(auto fd : temp)
                {
                    close(fd);
                }
                PrintFd(temp);
            }
            close(pipefd[1]);
            dup2(pipefd[0],0);//将管道文件 重定向 到 标准输入中, 管道文件就是输入
            Work();
            exit(0);
        }

        //3.2 父进程 关闭读
        close(pipefd[0]);   //这样并不知道 具体管道 对应 哪个子进程
        temp.push_back(pipefd[1]);//
        //对每个管道命名管理
        c->push_back(channel(pipefd[1],id));//push_back只接受一个参数，只有单参数自定义类型才会隐式类型转换
    }
}

对于 Work() 函数，即子进程执行任务的函数，我们父进程发送一个4字节的整数，判断执行不同的任务。

void Work()
{
    while(true)
    {
        int code = 0;
        //可以直接发送整形，即二进制，也可以显示使用“”发送字符串，再stoi
        int n = read(0,&code,sizeof(code));//从fd=0中读，每次期望读4个字节, 判断执行什么任务
       if(n == sizeof(code))//要读到4个字节
       {
           if (!init.CheckSafe(code)) // code时读到的值，n是读到的个数
               continue;
           init.RunTask(code); // 执行对应任务
       }
       else if(n == 0)
       {
            break;
       }
       
    }
    std::cout << "child quit"<<std::endl;
}

关于read，默认是阻塞模式，当写端关闭才会返回0，这里如果父进程退出，曾经为每一个子进程创建的写端描述符就关闭了，read就会返回0，表示读到0字节数据，读到文件结尾，然后自己进程判断后也会关闭，所以我们可以通过主动关闭写端可以控制子进程退出。

场景	阻塞模式（默认）	非阻塞模式（O_NONBLOCK）
管道/套接字缓冲区为空，但对端写端仍打开	read 阻塞，不返回，直到有数据	立即返回 -1，errno 设为 EAGAIN 或 EWOULDBLOCK
对端写端已关闭（EOF）	返回 0（表示没有更多数据）	返回 0

4. 发送消息

我们这里选择轮询管道，随机任务。flag = true 用来表示一直发送信息给子进程。

void SendCommand(const std::vector<channel> &c, bool flag, int num = -1)//随机任务，轮询管道使用
{
    int pos = 0;
    while(true)
    {
        // 1.选择任务
        int command = init.SelectTask();
        // 2.选择信道
        const channel &p = c[pos++];//c是指针时，不是引用，所以c[pos]等价于*(c+pos) 还是vector<channel>类型
        pos %= c.size();

        //debug
        std::cout << "send command " << init.ToDes(command) << "[" << command << "]" << " in " << p.name << " worker is: "<< p.workerid << std::endl;

        // 3.发送任务
        write(p.ctrlfd, &command, sizeof(command));
        
        //4.判断是否退出
        if(flag == false)
        {
            num--;
            if(num == 0)
                break;
        }

        sleep(1);
    }

    std::cout << "send done" << std::endl;
}

5. 回收管道

这里讲一下为什么需要使用 std::vector<int> temp 记录一下父进程创建曾经创建的子进程。父进程在创建子进程时，子进程会复制父进程的文件描述符表，所以如果不关闭父进程之前创建的管道文件对应的文件描述符，之后创建的子进程也可以看到之前父进程创建的管道，效果如下图所示：channel1 有 3 个子进程可以写，channel2有两个子进程可以写。

当以这种方式创建多个子进程时，默认情况下父进程按顺序从上到下关闭写端时，read是不返回0的，所以需要先把父进程的管道全部关闭，然后子进程自底向上退出。

void ReleaseChannels(const std::vector<channel> channels)
{
    //verson1:回收资源,想让子进程退出，并且释放管道，只要关闭写端
    for(const auto &c:channels)
    {
        close(c.ctrlfd);
    }
    for(const auto &c:channels)
    {
        //构建channel是保存子进程pid就是方便这里回收
        //这里阻塞等待，都会被等待到，但实际还是子进程先变成僵尸进程
        pid_t rid = waitpid(c.workerid,nullptr,0);
        if(rid == c.workerid)
        {
            std::cout << "wait child " << c.workerid << std::endl;
        }
    }
}
void ReleaseChannels(const std::vector<channel> channels)
{
    //version2: 倒状回收,先关闭最后一个
    int num = channels.size()-1;
    for(;num>=0;num--)
    {
        close(channels[num].ctrlfd);
        waitpid(channels[num].workerid,nullptr,0);
    }
}   

我们文章第三部分创建信道时，子进程已经关闭多余的文件描述符，可以直接 version1 顺序释放。如果不关闭多余文件描述符，直接version1 顺序回收，会阻塞。

注意：读写操作都是原子操作，不用担心正在写时数据的一半被读取，当一端操作时另一端必须等待，上一篇文章提到管道大小64kb，也不用担心写数据写不完。

本篇结束！