Linux 管道通信实战：匿名管道进程池与命名管道服务端模型

3. 命名管道

匿名管道只能用在有血缘关系的进程之间。如果我们想用在不相关的进程之间通信，可以使用命名管道（FIFO）完成！无关的进程之间想要通信，必须看到同一份资源，所以命名管道必须有路径（名字），双方才都能看到他。

命名管道本质是一种特殊类型的文件——管道文件。

命名管道可以从命令行上创建，使用命令 mkfifo 文件名。

也可以在程序中创建，使用函数 mkfifo，第一个参数是文件名，第二个参数是文件权限。成功创建返回 0，失败返回 -1。

命名管道使用完需要我们手动删除，可以使用函数 unlink 删除文件的方式。

匿名管道由 pipe 函数创建并打开；命名管道由 mkfifo 函数创建，用 open 打开。它们的唯一区别在于创建与打开的方式不同，这些工作完成时候，它们具有相同的语义。

三、实例：匿名管道实现进程池

进程池，是指提前创建好多个子进程，在需要使用时直接分配任务。省去了创建子进程的开销。

父进程需要管理'通道'，组织管理子进程。

#include <cassert>
#include <cstdio>
#include <cstdlib>
#include <ctime>
#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <vector>

// 进程池，是指提前创建好多个子进程，在需要使用时直接分配任务。省去了创建子进程的开销
// 父进程需要管理'通道'，组织管理子进程
enum {
    OK = 0,
    PIPE_ERR,
    FORK_ERR,
    READ_ERR,
    WRITE_ERR,
    WAIT_ERR
};

// 全局定义好子进程数量，任务数量
const int gprocessnum = 7;

void task1() {
    std::cout << "这是下载数据任务" << std::endl;
}

void task2() {
    std::cout << "这是打印日志任务" << std::endl;
}

void task3() {
    std::cout << "这是刷新磁盘任务" << std::endl;
}

void task4() {
    std::cout << "这是更新用户状态任务" << std::endl;
}

typedef void (*task_t)();

// 任务表
const int gtasknum = 4;
task_t tasks[gtasknum] = {task1, task2, task3, task4};

class ProcessPool {
private:
    // 内部类维护子进程通道
    class Channel {
    private:
        int _wfd;     // 当前子进程通道的管道写端 fd
        pid_t _id;    // 子进程 id
        std::string channel_name; // 自定义通道的名字

    public:
        Channel(int wfd, pid_t id) : _wfd(wfd), _id(id) {
            channel_name = "channel-" + std::to_string(id);
        }

        void ClosePipe() {
            close(_wfd);
        }

        void PrintInfo() {
            printf("管道 wfd: %d, 通道名：%s\n", _wfd, channel_name.c_str());
        }

        int getfd() {
            return _wfd;
        }

        const char* getname() {
            return channel_name.c_str();
        }

        void Wait() {
            pid_t rid = waitpid(_id, nullptr, 0);
            if (rid < 0) {
                // wait 出错
                exit(WAIT_ERR);
            }
            std::cout << "回收子进程：" << _id << std::endl;
        }
    };

public:
    ProcessPool() {
        // 种下随机数种子
        srand(time(NULL));
    }

    // 初始化进程池，创建好若干个子进程通道
    void Init() {
        CreateProcessChannels();
    }

    // 打印通道信息验证
    void Debug() {
        for (auto& channel : channels) {
            channel.PrintInfo();
        }
    }

    // 随机分配任务执行
    void Run() {
        int cnt = 10;
        while (cnt--) {
            int itask = SelectTask();
            int ichannel = SelectChannel();
            printf("向通道%s发送任务 task%d...\n", channels[ichannel].getname(), itask + 1);
            SendTask2Channel(itask, ichannel);
            sleep(1);
        }
    }

    void Quit() {
        for (auto& channel : channels) {
            channel.ClosePipe();
            channel.Wait();
        }
    }

private:
    void CreateProcessChannels() {
        for (int i = 0; i < gprocessnum; i++) {
            int pipefd[2] = {0};
            int n = pipe(pipefd);
            if (n < 0) {
                // 管道创建出错
                exit(PIPE_ERR);
            }
            pid_t id = fork();
            if (id < 0) {
                // 子进程创建出错
                exit(FORK_ERR);
            } else if (id == 0) {
                // 子进程 read
                // 子进程的文件描述符表是拷贝父进程的。父进程 fd 表中有指向其他管道的 wfd，子进程必须关闭指向其他管道的 wfd！
                if (!channels.empty()) {
                    for (auto& channel : channels)
                        channel.ClosePipe();
                }
                // 子进程从管道中读数据，关闭写端
                close(pipefd[1]);
                // 子进程进入待执行任务状态，将来执行完成后回来退出
                DoTask(pipefd[0]);
                exit(OK);
            } else {
                // 父进程 write
                // 关闭管道读端
                close(pipefd[0]);
                channels.emplace_back(pipefd[1], id);
                printf("创建子进程%d成功\n", id);
            }
        }
    }

    void DoTask(int fd) {
        // 子进程需要持续监听管道，等待父进程下发任务，直到父进程主动关闭管道写端。
        while (1) {
            int task_code;
            ssize_t n = read(fd, &task_code, sizeof(task_code));
            if (n == sizeof(task_code)) {
                // 根据读取到的 task_code 从任务表中选择函数执行
                if (task_code >= 0 && task_code < gtasknum) {
                    tasks[task_code]();
                }
            } else if (n == 0) {
                // 读到了文件尾，说明管道写端关闭了
                printf("%d任务退出\n", getpid());
                break;
            } else {
                // read 出错
                exit(READ_ERR);
            }
        }
    }

    int SelectTask() {
        // 随机选一个任务
        return rand() % gtasknum;
    }

    int SelectChannel() {
        // 依次选择子进程
        static int i = 0;
        int selected = i;
        i++;
        i %= gprocessnum;
        return selected;
    }

    void SendTask2Channel(int itask, int ichannel) {
        assert(0 <= itask && itask < gtasknum && ichannel >= 0 && ichannel < gprocessnum);
        ssize_t n = write(channels[ichannel].getfd(), &itask, sizeof(itask));
        if (n < 0) {
            // write 出错
            exit(WRITE_ERR);
        }
    }

private:
    // 组织所有的子进程通道
    std::vector<Channel> channels;
};

int main() {
    ProcessPool pp;
    // 初始化进程池，创建好若干个子进程通道
    pp.Init();
    // 打印通道信息验证
    pp.Debug();
    // 随机分配任务执行
    pp.Run();
    // 释放管道，回收子进程
    pp.Quit();
    return 0;
}

四、实例：命名管道实现服务端客户端通信模型

这里我们封装了一个 Fifo 类来简化操作，包含创建、打开、删除管道以及收发消息的功能。

// Fifo.hpp
#pragma once
#include <cstdio>
#include <cstring>
#include <fcntl.h>
#include <iostream>
#include <string>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#define FORREAD 1
#define FORWRITE 2

const std::string myfifo = "./fifo";

class Fifo {
public:
    Fifo(const std::string& filename = myfifo) : _filename(filename), _mode(0666), _fd(-1) {}

    // 创建管道
    void Build() {
        // 如果管道文件已存在，就 return
        if (IsExist())
            return;
        int n = mkfifo(_filename.c_str(), _mode);
        if (n < 0) {
            std::cerr << "mkfifo error: " << strerror(errno) << std::endl;
            exit(1);
        }
        std::cout << "mkfifo success" << std::endl;
    }

    // 打开管道
    void Open(int mode) {
        if (mode == FORREAD) {
            _fd = open(_filename.c_str(), O_RDONLY);
        } else if (mode == FORWRITE) {
            _fd = open(_filename.c_str(), O_WRONLY);
        }
        if (_fd < 0) {
            std::cerr << "open error: " << strerror(errno) << std::endl;
            exit(2);
        }
        std::cout << "open success" << std::endl;
    }

    // 删除管道
    void Delete() {
        if (!IsExist()) {
            return;
        }
        int n = unlink(_filename.c_str());
        if (n < 0) {
            std::cerr << "delete error: " << strerror(errno) << std::endl;
            exit(3);
        }
        std::cout << "delete success" << std::endl;
    }

    // 发送消息
    void Send(std::string& msgin) {
        ssize_t n = write(_fd, msgin.c_str(), msgin.size());
    }

    // 接受消息
    int Receive(std::string& msgout) {
        char buffer[128];
        ssize_t n = read(_fd, buffer, sizeof(buffer) - 1);
        if (n > 0) {
            buffer[n] = '\0';
            msgout = buffer;
            return n;
        } else if (n == 0) {
            return 0;
        } else {
            return -1;
        }
    }

private:
    bool IsExist() {
        struct stat st;
        // stat 函数用于查询一个文件的属性，如果查到了返回 0
        // 利用这一点判断管道文件是否存在
        int n = stat(_filename.c_str(), &st);
        if (n == 0) {
            return true;
        } else {
            errno = 0; // 消除这次失败对后面代码的影响
            return false;
        }
    }

private:
    std::string _filename;
    mode_t _mode;
    int _fd;
};

服务端代码：

// Server.cc
#include "Fifo.hpp"

int main() {
    // 服务端 创建并打开管道
    Fifo pipefile;
    pipefile.Build();
    pipefile.Open(FORREAD);
    std::string msg;
    while (1) {
        int n = pipefile.Receive(msg);
        if (n > 0) {
            std::cout << "客户端说：" << msg << std::endl;
        } else {
            break;
        }
    }
    pipefile.Delete();
    return 0;
}

客户端代码：

// Client.cc
#include "Fifo.hpp"

int main() {
    // 客户端 写入信息
    Fifo fileclient;
    fileclient.Open(FORWRITE);
    while (1) {
        std::cout << "请输入：" << std::endl;
        std::string msg;
        std::getline(std::cin, msg);
        fileclient.Send(msg);
    }
    return 0;
}

通过这两个实例，我们可以清晰地看到匿名管道在父子进程间的便捷性，以及命名管道如何打破进程血缘限制，实现更通用的通信架构。实际开发中，根据场景选择合适的 IPC 方式至关重要。