C++ 實現(xiàn)進程池：主從架構、管道通信與任務調度(示例詳解)

更新時間：2026年06月01日 10:24:33 作者：zincsweet

從進程池架構中,主進程負責任務分配與管理,子進程專注于任務處理,二者通過管道通信實現(xiàn)高效穩(wěn)定的任務執(zhí)行機制,本文詳細解析了進程池的創(chuàng)建、任務調度及退出流程,并強調了關閉歷史遺留文件描述符的重要性,感興趣的朋友一起看看吧

主/從進程池架構

一、原理

1 個主進程（Master）+ 預先創(chuàng)建 N 個子進程（Worker）

主進程只管分配任務，子進程只管處理任務，

通過管道通信，實現(xiàn)高并發(fā)、穩(wěn)定、不重復創(chuàng)建銷毀進程。

二、三大核心角色

1. Master（主進程）

唯一
不處理業(yè)務
只做 4 件事：
1. 創(chuàng)建進程池（fork 一堆子進程）
2. 接收任務
3. 分發(fā)任務（發(fā)給空閑子進程）
4. 管理子進程（監(jiān)控、重啟、回收）

2. Worker（子進程 → 進程池）

預先創(chuàng)建好 N 個
一直活著，不退出
阻塞等待任務
收到任務 → 處理 → 繼續(xù)等待

3. 管道（通信通道）

每個子進程一條管道
Master 寫 → 發(fā)任務
Worker 讀 → 收任務
阻塞等待，天然實現(xiàn)同步控制

三、代碼

#include <cstdlib>
#include <ctime>
#include <iostream>
#include <string>
#include <unistd.h>
#include <vector>
#include <functional>
#include <sys/wait.h>
#include <sys/types.h>
#define POLL_SIZE 4
// ***********************************任務列表*********************************************
void SyncDisk() {
    std::cout << "同步磁盤..." << std::endl;
}
void DownloadFile() {
    std::cout << "下載文件..." << std::endl;
}
void PrintMessage() {
    std::cout << "打印消息..." << std::endl;
}
void UpdateDatabase() {
    std::cout << "更新數(shù)據(jù)庫..." << std::endl;
}
typedef void (*task_t)();   // 定義一個函數(shù)指針類型，指向任務函數(shù)
task_t tasks[] = {SyncDisk, DownloadFile, PrintMessage, UpdateDatabase};  // 任務列表
// ***********************************進程池實現(xiàn)*******************************************
enum {
    OK = 0,
    PIPE_ERROR,
    FORK_ERROR
};
void Dotask(int fd) {   // 任務的入口
    while (1) {
        int task_code = 0;
        ssize_t n = read(fd, &task_code, sizeof(task_code));    // 約定好讀取4字節(jié)的任務碼
        if (n == -1) {
            std::cerr << "read error" << std::endl;
            break;
        }
        else if (n == 0) {
            std::cout << "沒有任務了..." << std::endl;
            break; // 管道被關閉，退出循環(huán)
        }
        else {
            if (task_code < 0 || task_code >= sizeof(tasks) / sizeof(task_t)) {
                std::cerr << "Invalid task code: " << task_code << std::endl;
                continue; // 跳過無效的任務碼
            }
            // 根據(jù)任務碼執(zhí)行對應的任務函數(shù)
            tasks[task_code]();
        }
    }
}
// typedef std::function<void (int)> cb_t;
using cb_t = std::function<void (int)>;
class ProcessPool {
private:
    // 定義一個內部Channel類，保存管道寫端和子進程pid
    class Channel {
    public:
        Channel(int wfd, pid_t sub_pid)
            : _wfd(wfd), _sub_pid(sub_pid)
        {
            _sub_name = "sub_channel_" + std::to_string(sub_pid);
        }
        ~Channel()
        {
        }
        void Write(int task_index) {
            ssize_t n = write(_wfd, &task_index, sizeof(task_index));
            (void)n; // 忽略寫入結果
        }
        void PrintInfo() {
            printf("Channel Info - Sub PID: %d, Sub Name: %s\n", _sub_pid, _sub_name.c_str());
        }
        std::string GetSubName() const {
            return _sub_name;
        }
        void ClosePipe() {
            std::cout << "關閉管道wfd: " << _wfd << std::endl;
            close(_wfd);
        }
        void Wait() {
            waitpid(_sub_pid, nullptr, 0);
            std::cout << "回收子進程: " << " PID: " << _sub_pid << std::endl;
        }
    private:
        int _wfd;              // 寫管道文件描述符
        pid_t _sub_pid;        // 子進程pid
        std::string _sub_name; // 子進程名字
    };
public:
    ProcessPool() {
        srand((unsigned int)time(nullptr) ^ (unsigned int)getpid()); // 設置隨機數(shù)種子
    }
    ~ProcessPool(){}
    void Init(cb_t cb) {
        CreatProcessChannel(cb);
    }
    void Debug() {
        for (auto& ch : channels) {
            ch.PrintInfo();
        }
    }
    void Run() {
        int cnt = 6;
        while (cnt--) {
            std::cout << "---------------------------------------------" << std::endl;
            // 1. 選擇一個任務
            int itask = SelectTask();
            std::cout << "選擇任務： index: " << itask << std::endl;
            // 2. 選擇一個channel(管道+子進程)，本質是選擇一個下標
            int index = SelectChannel();
            std::cout << "選擇管道： index: " << index << std::endl;
            // 3. 發(fā)送任務給指定的channel(管道+子進程)
            SendTaskToChannel(itask, index);
            std::cout << "發(fā)送任務 index " << itask << " 到管道 index " << index << std::endl;
        }
    }
    void Quit() {
        // 想要1 : 1地回收，就要改變CreateProcessChannel的實現(xiàn)
        // 就要在子進程中關閉歷史遺留的管道寫端
        // 從而實現(xiàn)1 : 1地回收資源。
        for (auto& ch : channels) {
            ch.ClosePipe(); // 關閉寫端
            ch.Wait();
        }
        // 逆向回收資源
        // int end = channels.size() - 1;
        // for (int i = end; i >= 0; --i) {
        //     channels[i].ClosePipe(); // 關閉寫端
        //     channels[i].Wait();      // 等待子進程退出
        // }
        // 1:1回收演示
        // for (auto& ch : channels) {
        //     ch.ClosePipe(); // 關閉寫端
        //     ch.Wait();
        // }
        // // 1. 關閉所有管道寫端，通知子進程退出
        // for (auto& ch : channels) {
        //     ch.ClosePipe(); // 關閉寫端
        // }
        // // 2. 等待所有子進程退出
        // for (auto& ch : channels) {
        //     ch.Wait();
        // }
    }
private:
    int SelectChannel() {
        // 這里簡單的輪詢選擇一個channel
        static int index = 0;
        int selected = index % channels.size();
        ++index;
        return selected;
    }
    int SelectTask() {
        // 輪詢選擇一個任務
        int itask = rand() % (sizeof(tasks) / sizeof(task_t));
        return itask;
    }
    void SendTaskToChannel(int itask, int index) {
        if (itask < 0 || itask >= sizeof(tasks) / sizeof(task_t)) {
            std::cerr << "Invalid task index: " << itask << std::endl;
            return;
        }
        if (index < 0 || index >= channels.size()) {
            std::cerr << "Invalid channel index: " << index << std::endl;
            return;
        }
        // 將任務索引寫入管道，通知子進程執(zhí)行對應的任務
        channels[index].Write(itask);
    }
    void CreatProcessChannel(cb_t cb) {
        for (int i = 0; i < POLL_SIZE; ++i) {
            int pipefd[2] = {0};
            int ret = pipe(pipefd);
            if (ret == -1) {
                std::cerr << "pipe error" << std::endl;
                exit(PIPE_ERROR);
            }
            pid_t pid = fork();
            if (pid == -1) {
                std::cerr << "fork error" << std::endl;
                exit(FORK_ERROR);
            }
            else if (pid == 0) {
                // 子進程
                if (!channels.empty()) {
                    for (auto& ch : channels) {
                        ch.ClosePipe(); // 關閉歷史遺留的管道寫端
                    }
                }
                close(pipefd[1]);   // 關閉寫端
                cb(pipefd[0]);
                exit(OK);
            }
            else {
                // 父進程
                close(pipefd[0]);   // 關閉讀端
                // 創(chuàng)建一個channel對象，保存管道寫端和子進程pid
                channels.emplace_back(pipefd[1], pid);
                // Channel ch(pipefd[1], pid);
                // channels.emplace_back(ch);  // 將channel對象添加到容器中
                std::cout << "創(chuàng)建了一個管道PID: " << pid << "到管道容器中了" << std::endl;
                sleep(1);
            }
        }
    }
private:
    std::vector<Channel> channels;    // 要有未來組織所有channel的容器
};
int main() {
    // 1. 初始化一個進程池對象
    ProcessPool pool;
    pool.Init(Dotask);
    // 2. 運行進程池
    pool.Run();
    // 3. 結束進程池釋放資源
    pool.Quit();
    return 0;
}

Linux 進程池（ProcessPool）源碼解析

一、項目目標

實現(xiàn)一個簡單的進程池：

父進程負責調度任務
子進程負責執(zhí)行任務
父子進程通過匿名管道通信
父進程發(fā)送任務碼
子進程根據(jù)任務碼執(zhí)行對應任務

二、整體架構

                     Parent(ProcessPool)
                              │
         ┌────────────────────┼────────────────────┐
         │                    │                    │
         │                    │                    │
      pipe0                pipe1                pipe2
         │                    │                    │
         ▼                    ▼                    ▼
      Child0               Child1               Child2
         │                    │                    │
      Dotask()             Dotask()             Dotask()
         │                    │                    │
         └────────────執(zhí)行任務────────────┘

父進程：

ProcessPool

負責：

創(chuàng)建子進程
創(chuàng)建管道
選擇任務
分發(fā)任務
回收子進程

三、任務系統(tǒng)

任務定義

void SyncDisk();
void DownloadFile();
void PrintMessage();
void UpdateDatabase();

任務表

typedef void (*task_t)();
task_t tasks[] =
{
    SyncDisk,
    DownloadFile,
    PrintMessage,
    UpdateDatabase
};

本質：

任務碼 → 函數(shù)地址

對應關系：

任務碼	任務
0	SyncDisk
1	DownloadFile
2	PrintMessage
3	UpdateDatabase

四、ProcessPool類設計

class ProcessPool
{
private:
    vector<Channel> channels;
};

保存所有：

管道 + 子進程

五、Channel設計

Channel是什么

一個Channel對應：

一個管道
+
一個子進程

類結構

class Channel
{
private:
    int _wfd;
    pid_t _sub_pid;
    string _sub_name;
};

成員說明

_wfd

int _wfd;

父進程保存：

管道寫端

用于發(fā)送任務。

_sub_pid

pid_t _sub_pid;

保存：

子進程PID

用于：

waitpid()

回收資源。

_sub_name

sub_channel_xxx

用于調試。

六、進程創(chuàng)建流程

CreateProcessChannel()

核心代碼：

pipe(pipefd);

fork();

創(chuàng)建過程

第一次循環(huán)

Parent
    │
    └── Child0

生成：

pipe0

第二次循環(huán)

Parent
    ├── Child0
    └── Child1

生成：

pipe1

第三次循環(huán)

Parent
    ├── Child0
    ├── Child1
    └── Child2

生成：

pipe2

最終：

Parent
│
├── Child0
├── Child1
├── Child2
└── Child3

七、為什么關閉歷史遺留寫端

問題

fork會繼承所有打開文件描述符。

例如：

pipe0

創(chuàng)建Child0后：

Child0
擁有：
pipe0[0]
pipe0[1]

關閉：

close(pipe0[1]);

剩：

pipe0[0]

繼續(xù)fork Child1：

Child1會繼承：

pipe0[1]
pipe1[1]

此時：

Child1
持有 pipe0 寫端

如果父進程關閉：

pipe0 寫端

Child0仍然讀不到EOF。

因為：

還有進程持有寫端

解決方案

for(auto &ch : channels)
{
    ch.ClosePipe();
}

關閉歷史遺留寫端。

最終關系：

pipe0 → Child0
pipe1 → Child1
pipe2 → Child2
pipe3 → Child3

形成：

1 Pipe
    ↓
1 Child

八、任務執(zhí)行流程

Dotask()

void Dotask(int fd)
{
    while(true)
    {
        read(fd,&task_code,sizeof(task_code));
    }
}

收到任務碼

例如：

表示：

PrintMessage();

執(zhí)行：

tasks[2]();

執(zhí)行流程：

父進程
   │
write(2)
   │
   ▼
pipe
   │
   ▼
子進程
   │
read()
   │
   ▼
tasks[2]()
   │
   ▼
PrintMessage()

九、任務調度策略

SelectTask()

rand() % 4

隨機任務：

0~3

SelectChannel()

static int index = 0;
selected =
index % channels.size();

執(zhí)行順序：

這種策略叫：

Round Robin
輪詢調度

十、運行流程圖

┌───────────┐
│  Run()    │
└─────┬─────┘
      │
      ▼
選擇任務
      │
      ▼
選擇Channel
      │
      ▼
Write(任務碼)
      │
      ▼
管道傳輸
      │
      ▼
子進程read()
      │
      ▼
執(zhí)行任務
      │
      ▼
繼續(xù)等待任務

十一、退出流程

父進程

ClosePipe();

關閉所有寫端。

子進程

read(...)

表示：

EOF

執(zhí)行：

break;

退出循環(huán)。

父進程回收

waitpid(pid,nullptr,0);

流程：

Parent
    │
close(wfd)
    │
    ▼
Child
read()==0
    │
    ▼
退出
    │
    ▼
waitpid()
    │
    ▼
回收完成

十二、源碼中的亮點

回調思想

using cb_t =
std::function<void(int)>;

初始化：

pool.Init(Dotask);

創(chuàng)建子進程后：

cb(pipefd[0]);

調用：

Dotask(pipefd[0]);

實現(xiàn)：

進程池框架
+
業(yè)務邏輯

解耦。

到此這篇關于C++ 實現(xiàn)進程池：主從架構、管道通信與任務調度(示例詳解)的文章就介紹到這了,更多相關C++ 進程池主從架構內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

深入C++拷貝構造函數(shù)的總結詳解
本篇文章是對C++中拷貝構造函數(shù)進行了總結與介紹。需要的朋友參考下
2013-05-05
.h和.cpp文件的區(qū)別(zt)詳細介紹
特別是對源文件和頭文件的概念，需要深入對它了解,本文將詳細介紹，需要了解的朋友可以參考下
2012-11-11
C++類中的常數(shù)據(jù)成員與靜態(tài)數(shù)據(jù)成員之間的區(qū)別
常數(shù)據(jù)成員是指在類中定義的不能修改其值的一些數(shù)據(jù)成員，類似于我們以前學過的常變量，雖然是變量，也有自己的地址，但是一經(jīng)賦初值，便不能再被修改
2013-10-10
C語言實現(xiàn)abs和fabs絕對值
這篇文章主要介紹了C語言實現(xiàn)abs和fabs絕對值，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2021-01-01
Qt實現(xiàn)簡單折線圖表
這篇文章主要為大家詳細介紹了Qt實現(xiàn)簡單折線圖表，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2022-06-06
C語言文件操作實現(xiàn)數(shù)據(jù)持久化(幫你快速了解文件操作函數(shù))
持久數(shù)據(jù)其實就是將數(shù)據(jù)保存到數(shù)據(jù)庫,下面這篇文章主要給大家介紹了關于C語言文件操作實現(xiàn)數(shù)據(jù)持久化(幫你快速了解文件操作函數(shù))的相關資料,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下
2022-11-11
C和指針小結(推薦)
下面小編就為大家?guī)硪黄狢和指針小結(推薦)。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考，一起跟隨小編過來看看吧
2016-06-06
VC實現(xiàn)Windows多顯示器編程的方法
這篇文章主要介紹了VC實現(xiàn)Windows多顯示器編程的方法,涉及VC獲取屏幕分辨率及顯示參數(shù)等技巧,具有一定參考借鑒價值,需要的朋友可以參考下
2015-10-10
詳解C++程序中定義struct結構體的方法
C++中同樣擁有C語言中的結構體,下面就來詳解C++程序中定義struct結構體的方法,需要的朋友可以參考下
2016-05-05
詳解C++中十六進制字符串轉數(shù)字（數(shù)值）
這篇文章主要介紹了詳解C++中十六進制字符串轉數(shù)字（數(shù)值）的相關資料,這里提供兩種實現(xiàn)方法，需要的朋友可以參考下
2017-08-08

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线