C++ 實(shí)現(xiàn)進(jìn)程池：主從架構(gòu)、管道通信與任務(wù)調(diào)度(示例詳解)
更新時(shí)間：2026年06月01日 10:24:33 作者：zincsweet
從進(jìn)程池架構(gòu)中,主進(jìn)程負(fù)責(zé)任務(wù)分配與管理,子進(jìn)程專注于任務(wù)處理,二者通過(guò)管道通信實(shí)現(xiàn)高效穩(wěn)定的任務(wù)執(zhí)行機(jī)制,本文詳細(xì)解析了進(jìn)程池的創(chuàng)建、任務(wù)調(diào)度及退出流程,并強(qiáng)調(diào)了關(guān)閉歷史遺留文件描述符的重要性,感興趣的朋友一起看看吧
主/從進(jìn)程池架構(gòu)
一、原理
1 個(gè)主進(jìn)程（Master）+ 預(yù)先創(chuàng)建 N 個(gè)子進(jìn)程（Worker）
主進(jìn)程只管分配任務(wù)，子進(jìn)程只管處理任務(wù)，
通過(guò)管道通信，實(shí)現(xiàn)高并發(fā)、穩(wěn)定、不重復(fù)創(chuàng)建銷毀進(jìn)程。
二、三大核心角色
1. Master（主進(jìn)程）
- 唯一
- 不處理業(yè)務(wù)
- 只做 4 件事：
- 創(chuàng)建進(jìn)程池（fork 一堆子進(jìn)程）
- 接收任務(wù)
- 分發(fā)任務(wù)（發(fā)給空閑子進(jìn)程）
- 管理子進(jìn)程（監(jiān)控、重啟、回收）
2. Worker（子進(jìn)程 → 進(jìn)程池）
- 預(yù)先創(chuàng)建好 N 個(gè)
- 一直活著，不退出
- 阻塞等待任務(wù)
- 收到任務(wù) → 處理 → 繼續(xù)等待
3. 管道（通信通道）
- 每個(gè)子進(jìn)程一條管道
- Master 寫(xiě) → 發(fā)任務(wù)
- Worker 讀 → 收任務(wù)
- 阻塞等待，天然實(shí)現(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
// ***********************************任務(wù)列表*********************************************
void SyncDisk() {
std::cout << "同步磁盤..." << std::endl;
}
void DownloadFile() {
std::cout << "下載文件..." << std::endl;
}
void PrintMessage() {
std::cout << "打印消息..." << std::endl;
}
void UpdateDatabase() {
std::cout << "更新數(shù)據(jù)庫(kù)..." << std::endl;
}
typedef void (*task_t)(); // 定義一個(gè)函數(shù)指針類型，指向任務(wù)函數(shù)
task_t tasks[] = {SyncDisk, DownloadFile, PrintMessage, UpdateDatabase}; // 任務(wù)列表
// ***********************************進(jìn)程池實(shí)現(xiàn)*******************************************
enum {
OK = 0,
PIPE_ERROR,
FORK_ERROR
};
void Dotask(int fd) { // 任務(wù)的入口
while (1) {
int task_code = 0;
ssize_t n = read(fd, &task_code, sizeof(task_code)); // 約定好讀取4字節(jié)的任務(wù)碼
if (n == -1) {
std::cerr << "read error" << std::endl;
break;
}
else if (n == 0) {
std::cout << "沒(méi)有任務(wù)了..." << std::endl;
break; // 管道被關(guān)閉，退出循環(huán)
}
else {
if (task_code < 0 || task_code >= sizeof(tasks) / sizeof(task_t)) {
std::cerr << "Invalid task code: " << task_code << std::endl;
continue; // 跳過(guò)無(wú)效的任務(wù)碼
}
// 根據(jù)任務(wù)碼執(zhí)行對(duì)應(yīng)的任務(wù)函數(shù)
tasks[task_code]();
}
}
}
// typedef std::function<void (int)> cb_t;
using cb_t = std::function<void (int)>;
class ProcessPool {
private:
// 定義一個(gè)內(nèi)部Channel類，保存管道寫(xiě)端和子進(jìn)程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; // 忽略寫(xiě)入結(jié)果
}
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 << "關(guān)閉管道wfd: " << _wfd << std::endl;
close(_wfd);
}
void Wait() {
waitpid(_sub_pid, nullptr, 0);
std::cout << "回收子進(jìn)程: " << " PID: " << _sub_pid << std::endl;
}
private:
int _wfd; // 寫(xiě)管道文件描述符
pid_t _sub_pid; // 子進(jìn)程pid
std::string _sub_name; // 子進(jìn)程名字
};
public:
ProcessPool() {
srand((unsigned int)time(nullptr) ^ (unsigned int)getpid()); // 設(shè)置隨機(jī)數(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. 選擇一個(gè)任務(wù)
int itask = SelectTask();
std::cout << "選擇任務(wù)： index: " << itask << std::endl;
// 2. 選擇一個(gè)channel(管道+子進(jìn)程)，本質(zhì)是選擇一個(gè)下標(biāo)
int index = SelectChannel();
std::cout << "選擇管道： index: " << index << std::endl;
// 3. 發(fā)送任務(wù)給指定的channel(管道+子進(jìn)程)
SendTaskToChannel(itask, index);
std::cout << "發(fā)送任務(wù) index " << itask << " 到管道 index " << index << std::endl;
}
}
void Quit() {
// 想要1 : 1地回收，就要改變CreateProcessChannel的實(shí)現(xiàn)
// 就要在子進(jìn)程中關(guān)閉歷史遺留的管道寫(xiě)端
// 從而實(shí)現(xiàn)1 : 1地回收資源。
for (auto& ch : channels) {
ch.ClosePipe(); // 關(guān)閉寫(xiě)端
ch.Wait();
}
// 逆向回收資源
// int end = channels.size() - 1;
// for (int i = end; i >= 0; --i) {
// channels[i].ClosePipe(); // 關(guān)閉寫(xiě)端
// channels[i].Wait(); // 等待子進(jìn)程退出
// }
// 1:1回收演示
// for (auto& ch : channels) {
// ch.ClosePipe(); // 關(guān)閉寫(xiě)端
// ch.Wait();
// }
// // 1. 關(guān)閉所有管道寫(xiě)端，通知子進(jìn)程退出
// for (auto& ch : channels) {
// ch.ClosePipe(); // 關(guān)閉寫(xiě)端
// }
// // 2. 等待所有子進(jìn)程退出
// for (auto& ch : channels) {
// ch.Wait();
// }
}
private:
int SelectChannel() {
// 這里簡(jiǎn)單的輪詢選擇一個(gè)channel
static int index = 0;
int selected = index % channels.size();
++index;
return selected;
}
int SelectTask() {
// 輪詢選擇一個(gè)任務(wù)
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;
}
// 將任務(wù)索引寫(xiě)入管道，通知子進(jìn)程執(zhí)行對(duì)應(yīng)的任務(wù)
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) {
// 子進(jìn)程
if (!channels.empty()) {
for (auto& ch : channels) {
ch.ClosePipe(); // 關(guān)閉歷史遺留的管道寫(xiě)端
}
}
close(pipefd[1]); // 關(guān)閉寫(xiě)端
cb(pipefd[0]);
exit(OK);
}
else {
// 父進(jìn)程
close(pipefd[0]); // 關(guān)閉讀端
// 創(chuàng)建一個(gè)channel對(duì)象，保存管道寫(xiě)端和子進(jìn)程pid
channels.emplace_back(pipefd[1], pid);
// Channel ch(pipefd[1], pid);
// channels.emplace_back(ch); // 將channel對(duì)象添加到容器中
std::cout << "創(chuàng)建了一個(gè)管道PID: " << pid << "到管道容器中了" << std::endl;
sleep(1);
}
}
}
private:
std::vector<Channel> channels; // 要有未來(lái)組織所有channel的容器
};
int main() {
// 1. 初始化一個(gè)進(jìn)程池對(duì)象
ProcessPool pool;
pool.Init(Dotask);
// 2. 運(yùn)行進(jìn)程池
pool.Run();
// 3. 結(jié)束進(jìn)程池釋放資源
pool.Quit();
return 0;
}Linux 進(jìn)程池（ProcessPool）源碼解析
一、項(xiàng)目目標(biāo)
實(shí)現(xiàn)一個(gè)簡(jiǎn)單的進(jìn)程池：
- 父進(jìn)程負(fù)責(zé)調(diào)度任務(wù)
- 子進(jìn)程負(fù)責(zé)執(zhí)行任務(wù)
- 父子進(jìn)程通過(guò)匿名管道通信
- 父進(jìn)程發(fā)送任務(wù)碼
- 子進(jìn)程根據(jù)任務(wù)碼執(zhí)行對(duì)應(yīng)任務(wù)
二、整體架構(gòu)
Parent(ProcessPool)
│
┌────────────────────┼────────────────────┐
│ │ │
│ │ │
pipe0 pipe1 pipe2
│ │ │
▼ ▼ ▼
Child0 Child1 Child2
│ │ │
Dotask() Dotask() Dotask()
│ │ │
└────────────執(zhí)行任務(wù)────────────┘父進(jìn)程：
ProcessPool
負(fù)責(zé)：
- 創(chuàng)建子進(jìn)程
- 創(chuàng)建管道
- 選擇任務(wù)
- 分發(fā)任務(wù)
- 回收子進(jìn)程
三、任務(wù)系統(tǒng)
任務(wù)定義
void SyncDisk(); void DownloadFile(); void PrintMessage(); void UpdateDatabase();
任務(wù)表
typedef void (*task_t)();
task_t tasks[] =
{
SyncDisk,
DownloadFile,
PrintMessage,
UpdateDatabase
};本質(zhì)：
任務(wù)碼 → 函數(shù)地址
對(duì)應(yīng)關(guān)系：
| 任務(wù)碼 | 任務(wù) |
|---|---|
| 0 | SyncDisk |
| 1 | DownloadFile |
| 2 | PrintMessage |
| 3 | UpdateDatabase |
四、ProcessPool類設(shè)計(jì)
class ProcessPool
{
private:
vector<Channel> channels;
};
保存所有：
管道 + 子進(jìn)程
五、Channel設(shè)計(jì)
Channel是什么
一個(gè)Channel對(duì)應(yīng)：
一個(gè)管道 + 一個(gè)子進(jìn)程
類結(jié)構(gòu)
class Channel
{
private:
int _wfd;
pid_t _sub_pid;
string _sub_name;
};
成員說(shuō)明
_wfd
int _wfd;
父進(jìn)程保存：
管道寫(xiě)端
用于發(fā)送任務(wù)。
_sub_pid
pid_t _sub_pid;
保存：
子進(jìn)程PID
用于：
waitpid()
回收資源。
_sub_name
sub_channel_xxx
用于調(diào)試。
六、進(jìn)程創(chuàng)建流程
CreateProcessChannel()
核心代碼：
pipe(pipefd); fork();
創(chuàng)建過(guò)程
第一次循環(huán)
Parent
│
└── Child0
生成：
pipe0
第二次循環(huán)
Parent
├── Child0
└── Child1
生成：
pipe1
第三次循環(huán)
Parent
├── Child0
├── Child1
└── Child2
生成：
pipe2
最終：
Parent │ ├── Child0 ├── Child1 ├── Child2 └── Child3
七、為什么關(guān)閉歷史遺留寫(xiě)端
問(wèn)題
fork會(huì)繼承所有打開(kāi)文件描述符。
例如：
pipe0
創(chuàng)建Child0后：
Child0 擁有： pipe0[0] pipe0[1]
關(guān)閉：
close(pipe0[1]);
剩：
pipe0[0]
繼續(xù)fork Child1：
Child1會(huì)繼承：
pipe0[1] pipe1[1]
此時(shí)：
Child1 持有 pipe0 寫(xiě)端
如果父進(jìn)程關(guān)閉：
pipe0 寫(xiě)端
Child0仍然讀不到EOF。
因?yàn)椋?/p>
還有進(jìn)程持有寫(xiě)端
解決方案
for(auto &ch : channels)
{
ch.ClosePipe();
}
關(guān)閉歷史遺留寫(xiě)端。
最終關(guān)系：
pipe0 → Child0 pipe1 → Child1 pipe2 → Child2 pipe3 → Child3
形成：
1 Pipe
↓
1 Child
八、任務(wù)執(zhí)行流程
Dotask()
void Dotask(int fd)
{
while(true)
{
read(fd,&task_code,sizeof(task_code));
}
}
收到任務(wù)碼
例如：
2
表示：
PrintMessage();
執(zhí)行：
tasks[2]();
執(zhí)行流程：
父進(jìn)程 │ write(2) │ ▼ pipe │ ▼ 子進(jìn)程 │ read() │ ▼ tasks[2]() │ ▼ PrintMessage()
九、任務(wù)調(diào)度策略
SelectTask()
rand() % 4
隨機(jī)任務(wù)：
0~3
SelectChannel()
static int index = 0; selected = index % channels.size();
執(zhí)行順序：
0 1 2 3 0 1 2 3 ...
這種策略叫：
Round Robin 輪詢調(diào)度
十、運(yùn)行流程圖
┌───────────┐
│ Run() │
└─────┬─────┘
│
▼
選擇任務(wù)
│
▼
選擇Channel
│
▼
Write(任務(wù)碼)
│
▼
管道傳輸
│
▼
子進(jìn)程read()
│
▼
執(zhí)行任務(wù)
│
▼
繼續(xù)等待任務(wù)
十一、退出流程
父進(jìn)程
ClosePipe();
關(guān)閉所有寫(xiě)端。
子進(jìn)程
read(...)
返回：
0
表示：
EOF
執(zhí)行：
break;
退出循環(huán)。
父進(jìn)程回收
waitpid(pid,nullptr,0);
流程：
Parent
│
close(wfd)
│
▼
Child
read()==0
│
▼
退出
│
▼
waitpid()
│
▼
回收完成
十二、源碼中的亮點(diǎn)
回調(diào)思想
using cb_t = std::function<void(int)>;
初始化：
pool.Init(Dotask);
創(chuàng)建子進(jìn)程后：
cb(pipefd[0]);
調(diào)用：
Dotask(pipefd[0]);
實(shí)現(xiàn)：
進(jìn)程池框架 + 業(yè)務(wù)邏輯
解耦。
到此這篇關(guān)于C++ 實(shí)現(xiàn)進(jìn)程池：主從架構(gòu)、管道通信與任務(wù)調(diào)度(示例詳解)的文章就介紹到這了,更多相關(guān)C++ 進(jìn)程池主從架構(gòu)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
深入C++拷貝構(gòu)造函數(shù)的總結(jié)詳解
本篇文章是對(duì)C++中拷貝構(gòu)造函數(shù)進(jìn)行了總結(jié)與介紹。需要的朋友參考下2013-05-05
特別是對(duì)源文件和頭文件的概念，需要深入對(duì)它了解,本文將詳細(xì)介紹，需要了解的朋友可以參考下2012-11-11
C++類中的常數(shù)據(jù)成員與靜態(tài)數(shù)據(jù)成員之間的區(qū)別
常數(shù)據(jù)成員是指在類中定義的不能修改其值的一些數(shù)據(jù)成員，類似于我們以前學(xué)過(guò)的常變量，雖然是變量，也有自己的地址，但是一經(jīng)賦初值，便不能再被修改2013-10-10
C語(yǔ)言實(shí)現(xiàn)abs和fabs絕對(duì)值
這篇文章主要介紹了C語(yǔ)言實(shí)現(xiàn)abs和fabs絕對(duì)值，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-01-01
這篇文章主要為大家詳細(xì)介紹了Qt實(shí)現(xiàn)簡(jiǎn)單折線圖表，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2022-06-06
C語(yǔ)言文件操作實(shí)現(xiàn)數(shù)據(jù)持久化(幫你快速了解文件操作函數(shù))
持久數(shù)據(jù)其實(shí)就是將數(shù)據(jù)保存到數(shù)據(jù)庫(kù),下面這篇文章主要給大家介紹了關(guān)于C語(yǔ)言文件操作實(shí)現(xiàn)數(shù)據(jù)持久化(幫你快速了解文件操作函數(shù))的相關(guān)資料,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2022-11-11
下面小編就為大家?guī)?lái)一篇C和指針小結(jié)(推薦)。小編覺(jué)得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考，一起跟隨小編過(guò)來(lái)看看吧2016-06-06
VC實(shí)現(xiàn)Windows多顯示器編程的方法
這篇文章主要介紹了VC實(shí)現(xiàn)Windows多顯示器編程的方法,涉及VC獲取屏幕分辨率及顯示參數(shù)等技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-10-10
詳解C++程序中定義struct結(jié)構(gòu)體的方法
C++中同樣擁有C語(yǔ)言中的結(jié)構(gòu)體,下面就來(lái)詳解C++程序中定義struct結(jié)構(gòu)體的方法,需要的朋友可以參考下2016-05-05
詳解C++中十六進(jìn)制字符串轉(zhuǎn)數(shù)字（數(shù)值）
這篇文章主要介紹了詳解C++中十六進(jìn)制字符串轉(zhuǎn)數(shù)字（數(shù)值）的相關(guān)資料,這里提供兩種實(shí)現(xiàn)方法，需要的朋友可以參考下2017-08-08