【Linux实践系列】：用c/c++制作一个简易的进程池

2025-04-27 22:00:18 0 阅读

🔥 本文专栏：Linux Linux实践项目
🌸作者主页：努力努力再努力wz

💪 今日博客励志语录： 人生没有标准答案，你的错题本也能写成传奇。

★★★ 本文前置知识：

匿名管道

1.前置知识回顾（对此十分熟悉的读者可以跳过）

那么在上一篇博客中，我们知道了进程之间具有通信的需求，因为进程之间需要合作协同完成某项任务，那么就需要各个进程之间进行分工合作，那么进程就需要知道对方完成的进度以及完成的结果，所以进程之间需要通信，但是进程无法直接访问对方的数据，因为进程之间具有独立性，所以为了达到进程的通信的需求又保证进程之间的独立性，那么操作系统采取的策略就是在内存中创建一块公共区域，那么一个进程向这个公共区域中写入，另一个进程从该公共区域读取，就能完成进程的通信

而对于父子进程或者说有血缘关系的进程，那么我们知道创建子进程的过程会拷贝父进程的task_struct结构体，并且修改其中的部分属性得到子进程自己独立的一份task_struct结构体，那么其中就会涉及到文件描述表的拷贝，那么意味着子进程会继承父进程打开的文件，而进程之间通信的核心思想便是创建一个公共区域，而由于子进程和父进程会共享被打开的文件，那么意味着文件就可以作为这个公共区域，所以父子进程通信的方式就是通过文件，所以在创建子进程之前，那么父进程会先创建一份用来通信的文件，而该文件不需要刷新写入到磁盘当中，因为该文件的内容只是临时用来保存父子之间写入的内容，不需要刷新到磁盘长时间来保存，所以需要创建一份内存级别也就是不需要刷新到磁盘的文件，那么其中就要调用pipe接口，那么它会创建两个分别以只读权限打开以及只写权限打开同一个管道文件的file结构体对象，并返回这两个结构体的文件描述符，然后再调用fork接口创建子进程，那么子进程会继承父进程创建的两个以不同权限打开的file结构体对象，而该文件只能单向通信，也就是只能一个进程往该文件中写入，另一个进程从该文件中读取，不能双方同时写入，不然会造成内容混乱，而正是由于一个进程只能往该文件写，另一个文件只能从该文件读，那么这个特点和我们生活中的自来水管道是十分相似的，因为自来水管道只能从一端流入，然后从一端流出，所以该文件又称之为管道文件，那么为了实现单向通信，就需要父子进程关闭各自其中的一个读写端

那么这就是对上文的内容大致回顾，如果你对此感到陌生，那么可以去看我上一期文章

进程池项目介绍

1.进程池的意义

那么这里我们用之前所学的内容来实现一个进程池，其中就包括匿名管道，那么首先在讲进程池具体实现之前，那么我们得知道进程池是用来干什么的，它有什么用，也就是做这个进程池有什么意义，那么想必这些问题是读者对于进程池首先的一个疑惑，所以这里我们就先来认识做进程池的意义

上文前置知识回顾的开篇我就说道过，进程之间需要共同来完成某项任务，那么此时进程就需要分工合作，来完成各自分配的任务，那么假设有这么一个场景，那么你现在有100个task要完成，然后你把这些task都准备交给子进程来完成，那么此时你是如何去分配这些任务给子进程呢？

那么有的小伙伴采取的是这种方式，也就是他先调用fork接口，然后创建一个子进程，然后给该子进程分配一个任务，接着父进程则是等待子进程退出，通过退出码来查看子进程完成的情况，如果子进程正常退出并且结果正确，那么接着它便继续调用fork接口重复上面的步骤，也就是循环创建子进程，然后给其分配任务让其执行，而对于父进程则是等待其子进程退出，获取其子进程的退出码，而现在有100个任务，那么意味着这个小伙伴要调用100次fork接口

而还有的小伙伴采取的是另一种方式，那么他不是创建出一个子进程，然后就直接给创建出的该子进程分配任务去完成，他则是先创建出一批的进程，比如20个进程，那么此时创建完这20个进程之后，那么此后他不会再调用fork接口去创建其他新的子进程，而就是利用手头上持有的这20个进程来完成这100个task，那么就需要父进程依次给这20个子进程分配各自的任务，然后分配完之后，等待这20个进程退出，获取其退出码，看进程是否正常退出，然后再依次给执行完任务结束的子进程继续分配新的任务

那么我们就来比较并且评价一下上面的这两个小伙伴各自的实现方式，首先明确的是，这两个小伙伴的实现方式肯定都是正确没有问题的，也就是说上面的这两种方式都能够成功的完成这100个task，但是这两种方式完成的效率就会有所差别，那么第二个小伙伴的实现的方式的效率要比第一个小伙伴的实现方式的效率要高很多，那么为什么呢？

那么首先我们一定要记住并且理解的一个道理那就是，系统接口的调用是具有代价的，虽然你在代码中for循环连续100次调用fork接口创建了一批子进程，然后一运行你的代码，发现程序还是正常运行并且结果正确，但是你要知道的是，fork系统调用接口底层所涉及到的工作，其中就包括会拷贝父进程的task_struct结构体然后修改其中的部分属性得到子进程自己独立的一份task_struct结构体，然后创建完子进程的task_struct结构体之后还涉及到写时拷贝以及页表的重新映射，并且操作系统还要将创建出来的子进程的task_struct结构体放到相应的队列中来维护管理比如放到就绪队列中，那么当子进程运行结束之后，那么又会涉及到子进程的task_struct结构体等各种资源的释放，那么从子进程的创建以及销毁所涉及到的工作就可以看出来，那么调用一个fork接口其实是有成本的，就如同以前你看到初中班上学习成绩十分优秀的同学，那么他上课的时候总是趴在桌子上睡觉，结果人家考试还次次考全班第一，你看着人家学习很轻松，但其实人家在你看不到的地方其实在偷偷努力，比如晚上学习到凌晨几点

所以对于第一种实现方式，那么它的缺点就是十分的明显，那么要多次调用系统接口，那么效率必然不会优秀，而第二种方式相比于第一种方式，那么它则是先创建一批子进程，俗话说磨刀不误砍柴功，那么这里我们先创建一批进程，但是不让其执行特定的任务，然后创建完之后，那么我们就只需要让这创建出的进程轮流去执行这100个任务

那么对于第一种方式，那么假设要交给子进程完成100个task，那么意味着要调用100次fork接口，然后这100个task就分别交给每一个fork创建出来的进程，最终完成这100个task，而对于第二种方式，假设我们预先创建20个进程，然后让这20个进程轮流执行完成这100个task任务，那么我们来对比一下这两种方式的效率

那么对于第一种，那么假设完成一个task的代价是k，那么调用fork接口的代价是m，那么第一种实现的方式的总代价就是100m+100K,而对于第二种方式来说，那么它预先创建了20个进程，来执行这100个task，那么对于第二种方式的总代价就是20m+100k,所以粗略估计下来，那么第二种方式明显比第一种更加优秀

而第二种方式正是我们进程池的实现的核心思想，那么为什么称其为进程池，我们就可以用和尚下山去取水的故事例子来理解：那么有一个和尚住在一个高山山顶上的一座寺庙，那么它如果要喝水或者洗澡只能到山脚下的小溪中去取水然后再将水运回山顶，那么一旦和尚口渴了或者想洗澡，那么意味着他就要跑到山脚下去取水，那么这样做明显代价就太大了并且十分的不划算，那么为了提升效率，减少上山下山的时间的浪费，那么和尚采取的做法就是在半山腰上建立一个蓄水池，先存储一大部分水，那么一旦有用水的需求就到这个池子中去即可，而不需要跑到山脚下去运水

所以我们为什么叫起进程“池”，那么这个池字就很形象，那么我们预先创建一批进程的这个过程和上文那个例子中建立一个蓄水池是一个道理，那么我们就不需要在去调用fork来去创建一个进程，直接从创建好的进程池中选取子进程去完成任务即可，那么这就是进程池的意义，目的就是为了提高效率，减少系统调用的开销

2.进程池的大体框架

那么知道了进程池的意义之后，那么我们再来说一下进程池的实现，那么首先我们脑海中得先有一个大体的实现框架以及思路，也就是说我们得先分析出进程池涉及到的各个模块，然后再来谈这各个模块具体的代码的实现

1.进程池的创建

那么根据上文，我们知道，那么我们在执行任务之前，首先得创建一批子进程，那么假设要创建的子进程的数量是n，那么意味着我们会涉及到一个循环，其中在循环内部调用n次fork接口来创建n个子进程，那么其次我们子进程是来完成某项任务的，那么这个任务的发送就得交给父进程，由父进程来分配给子进程要执行的任务，那么这个任务可以通过一个任务码来传递，也就是一个int类型的变量，那么既然父进程要给子进程发送任务码，那么必然就要涉及到进程之间的通信，而父子进程如何通信，我们也很熟悉了，那么便是通过匿名管道进行通信，所以这里就注意，在调用fork之前，那么我们得先调用pipe接口，所以刚才说的这一系列内容，比如管道以及子进程的创建，我们都可以把它封装到一个函数模块中，具体的实现细节下文会提到

2.任务列表的制作

那么我们知道子进程到时候是会通过管道读取父进程交给它的任务码，那么任务码的本质其实就是一个编号，因为到时候我们所有要执行的函数都会有一个函数指针指向它，那么最终会定义一个全局的函数指针数组，那么所谓的任务码就是对应着这个函数指针数组的一个下标，那么由于定义成了全局的指针数组，那么到时候fork创建子进程，那么子进程也能访问到这个函数指针数组，那么就可以读取管道中的任务码然后根据函数指针数组来执行相应的函数，那么我们要执行的各个任务的逻辑都是封装在函数当中，而我们函数指针数组就可以理解为任务列表，到时候我们就要完成函数指针数组的初始化，那么这个初始化工作就会交给一个函数来完成

3.子进程执行任务&&父进程传递任务

而我们知道我们会通过fork接口来创建子进程，然后利用fork的返回值使得父子进程有着不同的执行流，那么我们知道在创建子进程之前，会首先创建管道文件，那么接着调用fork创建子进程，那么意味着子进程会继承并且会和父进程共享着打开的管道文件，所以到时候在子进程的执行流中就需要关闭管道文件的写端，关闭完之后，下一步便是读取管道文件传来的任务码，获取到任务码然后执行任务，那么这就是子进程执行任务的大致思路，至于具体的细节，我们下文在进行补充

而父进程对应的代码段则是想管道文件中写入子进程要执行的任务码

4.资源的清理

那么资源的清理便是进程池的最后阶段了，那么这个阶段的工作就是父进程会关闭之前打开的管道文件，并且等待子进程退出，看子进程是否正常退出，那么具体的实现细节下文会说到

进程池的各个模块的具体实现

1.进程池的创建

那么这里我们进程池的创建专门放到process_init模块当中,那么其中会涉及到一个for循环的逻辑，然后在循环调用pipe接口，然后创建管道文件，得到管道文件的读写端的文件描述符，那么接着再调用fork接口创建子进程，然后利用fork的返回值，让父子进程有着各自的执行流，那么在子进程的执行流中，那么它会调用close接口来关闭管道文件的写端，而对于父进程则是关闭管道文件的读端

那么到时候父进程得要向管道文件中写入任务码，那么意味着父进程得知道管道文件的文件描述符，因为到时候向管道文件写入需要调用write接口，而write接口会接收一个文件描述符作为参数，向该文件描述符所指向的文件中写入一定字节数，并且我们还得知道该管道文件相连接的是哪个子进程，所以我们得记录子进程的PID，那么我们可以定义一个channel类，然后内部封装了两个成员变量，分别是管道文件的文件描述符以及其连接的子进程的PID，那么父进程在关闭玩对应的管道文件的读端之后，还要初始化channel对象，将其插入到一个vector数组中，那么vector数组中就维护了创建出来的各个管道的属性

std::vector<channel> channelarray;
class channel
{
    public:
     int _processid;
     int _write_fd;
     channel(int processid,int write_fd)
     :_processid(processid)
     ,_write_fd(write_fd)
     {
    }
};

而对于子进程来说，那么它关闭玩管道的写端之后，接着的任务就是去获取父进程在管道文件中写入的任务码以及执行任务，那么这个内容我们可以封装到一个start_mission函数模块中，那么我下文会详细解析这个函数

其次这里有一个小细节，那么到时候子进程要去管道文件读取任务码，那么这里我进行了一个重定向，也就是将子进程的管道文件重定向到标准输入文件，那么这里就会调用dup2接口，那么其会关闭标准输入文件，将标准输入文件的下标的指针指向管道文件，这样做的好处就是我们子进程在读取管道文件的输入的时候，不需要知道管道文件的文件描述符，统一的去标准输入的文件描述符中读取即可
 dup2(pipefd[0],0);
 close(pipefd[0]);

void processpool_init()
{
    for(int i=0;i<processnum;i++)
    {
      int pipefd[2];
       int n=pipe(pipefd);
       if(n<0)
       {
          perror("pipe fail");
          exit(EXIT_FAILURE);
       }
       int id=fork();
       if(id<0)
       {
       perror("fork");
       close(pipefd[0]);
       close(pipefd[1]);
       exit(EXIT_FAILURE);
       }
       if(id==0)
       {
           close(pipefd[1]);
           dup2(pipefd[0],0);
           close(pipefd[0]);
           start_mission();
           exit(0);
       }
      close(pipefd[0]);
      channelarray.push_back(channel(id,pipefd[1]));
    }
}

2.任务列表的制作

那么任务列表的制作就非常轻松，那么到时会我们会定义一个全局的函数指针数组，那么其中函数指针数组的每一个元素是一个函数指针指向一个函数，那么我们会将这个数组中的每一个元素给初始化指向对应的函数，那么这个函数就是子进程要执行的任务，那么函数指针数组的下标就是任务码，那么刚才所说的这些工作都交给mission_load来完成

#define missionnum 4
typedef void (*mission)() ;
std::vector<mission> missionarray;
void task1()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task1"<<std::endl;
}
void task2()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task2"<<std::endl;
}
void task3()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task3"<<std::endl;
}
void task4()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task4"<<std::endl;
}
void mission_load()
{
        missionarray.push_back(task1);
        missionarray.push_back(task2);
        missionarray.push_back(task3);
        missionarray.push_back(task4);
}

3.子进程执行任务&&父进程传递任务

那么子进程执行任务我们专门设置了一个start_mission函数模块来实现，那么其中在start_mission模块中，就会涉及到一个死循环，因为子进程不可能执行完一个任务就退出了，因为它还要继续被父进程分配执行下一个任务，就和之前实现shell外壳程序一样，那么整体的大框架也是一个死循环，那么你获取以及执行完用户输入的一个指令之后你的bash进程不可能就退出结束了吧，同理这里你子进程在获取父进程向管道文件中写入的任务码以及执行对应的函数之后，那么就循环继续读取下一次父进程向管道文件中的写入的任务码，所以涉及到一个死循环的逻辑

那么读取任务码就涉及到调用read接口，那么从上文可知我们已经将管道文件重定向到标准输入文件，那么这里我们就从标准输入文件中读取任务码，由于函数指针数组是全局变量，那么获取到任务码之后，直接根据函数指针数组执行相应的函数即可，而注意还要判断read的返回值，如果read返回0，说明了此时管道文件的写端已经被关闭，那么父进程已经关闭了该管道文件的写端，所以子进程没必要在进行读取，所以直接break，然后子进程退出

void start_mission()
{
     while(true)
     {
        int staues;
         int n=read(0,&staues,sizeof(int));
         if(n==sizeof(int))
         {
              if(staues>=0&&staues<missionnum)
              {
                std::cout<<"我是子进程"<<getpid()<<" 成功获取到任务码"<<staues<<std::endl;
                   missionarray[staues]();
              }
         }else if(n==0)
         {
            break;
         }else if(n<0)
         {
            perror("read");
            exit(EXIT_FAILURE);
         }

     }
}

而父进程要做的则是传递任务，我们同样也是定义一个process_control函数来实现，那么其中就要注意的就是负载均衡，所谓的负载均衡指的就是我们给创建出来的所有子进程分配任务的时候，希望让所有子进程都尽可能的分配执行到任务，也就是大家都能有事干，尽量别闲着，和操作系统调度进程是一个道理，那么做到负载均衡的方式有两种，第一种就是随机分配，那么由于之前我们用数组记录了每一个管道文件对应的channel对象，其中channel对象保存了子进程的编号，那么假设有n个管道，那么我们可以产生一个0到n-1的随机数，然后调用对应的子进程，由于产生0到n-1这每一个数的概率肯定是相等，所以可以做到负载均衡

其次第二种方式则是轮询，那么所谓的轮询就更加直观，就是我们先分配给任务按照子进程被创建的顺序依次分配，从第一个依次分配到最后一个，最后再回到第一个，那么其中就会涉及到取模运算

void process_control()
{
    srand((unsigned int)time(NULL));
	int which=0;
	for(int i=0;i<100;i++)
	{
		int cmd=rand()%missionnum;
		int n=write(channelarray[which]._write_fd,&cmd,sizeof(int));
		if(n<0)
		{
			perror("write");
			exit(EXIT_FAILURE);
		}
		std::cout<<"father process send a message to"<<channelarray[which]._processid<<" cmd :"<<cmd<<std::endl;
        
        which=(which+1)%processnum; 
	 } 
}

4.资源的清理

那么最后的资源清理任务则放到process_clean函数模块，那么这个模块就是关闭回收管道以及等待子进程，那么这里要注意的一点就是，我们每创建一个子进程，那么该子进程会继承之前创建出的所有管道文件，这会让管道文件的引用计数加一，那么子进程以及父进程会关闭各自的读写端，会让其引用计数减一，那么对于最后一个管道文件来说，那么它只被最后一个创建的子进程以及父进程所共享，那么由于子进程与父进程再关闭各自的读写端，那么最后一个管道文件的读写端的引用计数是1，那么以此往前类推，那么前面的管道文件的读写端就是从2开始递增，所以我们关闭管道文件得从最后一个管道文件往前关闭，不然你从前往后关闭的话，那么管道的引用计数不会为0，那么会导致资源泄漏并且子进程一直陷入阻塞状态，因为管道的写端未被关闭并且父进程一直没有写入

void process_clean()
{
    for(int i=l1.size()-1;i>=0;i--)
    {
          close(channelarray[i]._write_fd);
          int statues;
          int n=waitpid(channelarray[i]._processid,&statues,0);
          if(n<0)
          {
              perror("waitpid");
          }else
          {
              std::cout<<"子进程"<<channelarray[i]._processid<<"等待成功"<<std::endl;
          }
    }
}

完整实现

processpool.cpp

#include"processpool.hpp"
int main()
{
	mission_load();
	processpool_init();
	process_control();
    process_clean();
	 return 0;
}

processpool.hpp

include<iostream>
#include
#include
#include
#include
#include
#include
#include"task.hpp"
#define EXIT_FAILURE 1
#define missionnum 4
const int processnum=10;
std::vector<channel> channelarray;
class channel
{
    public:
     int _processid;
     int _write_fd;
     channel(int processid,int write_fd)
     :_processid(processid)
     ,_write_fd(write_fd)
     {
    }
};
void mission_load()
{
        missionarray.push_back(task1);
        missionarray.push_back(task2);
        missionarray.push_back(task3);
        missionarray.push_back(task4);
}
void start_mission()
{
     while(true)
     {
        int staues;
         int n=read(0,&staues,sizeof(int));
         if(n==sizeof(int))
         {
              if(staues>=0&&staues<missionnum)
              {
                std::cout<<"我是子进程"<<getpid()<<" 成功获取到任务码"<<staues<<std::endl;
                   missionarray[staues]();
              }
         }else if(n==0)
         {
            break;
         }else if(n<0)
         {
            perror("read");
            exit(EXIT_FAILURE);
         }

     }
}
void process_control()
{
    srand((unsigned int)time(NULL));
	int which=0;
	for(int i=0;i<100;i++)
	{
		int cmd=rand()%missionnum;
		int n=write(channelarray[which]._write_fd,&cmd,sizeof(int));
		if(n<0)
		{
			perror("write");
			exit(EXIT_FAILURE);
		}
		std::cout<<"father process send a message to"<<channelarray[which]._processid<<" cmd :"<<cmd<<std::endl;
        
        which=(which+1)%processnum; 
	 } 
}
void process_clean()
{
    for(int i=l1.size()-1;i>=0;i--)
    {
          close(channelarray[i]._write_fd);
          int statues;
          int n=waitpid(channelarray[i]._processid,&statues,0);
          if(n<0)
          {
              perror("waitpid");
          }else
          {
              std::cout<<"子进程"<<channelarray[i]._processid<<"等待成功"<<std::endl;
          }
    }
}
void processpool_init()
{
    for(int i=0;i<processnum;i++)
    {
      int pipefd[2];
       int n=pipe(pipefd);
       if(n<0)
       {
          perror("pipe fail");
          exit(EXIT_FAILURE);
       }
       int id=fork();
       if(id<0)
       {
       perror("fork");
       close(pipefd[0]);
       close(pipefd[1]);
       exit(EXIT_FAILURE);
       }
       if(id==0)
       {
           close(pipefd[1]);
           dup2(pipefd[0],0);
           start_mission();
           exit(0);
       }
      close(pipefd[0]);
      channelarray.push_back(channel(id,pipefd[1]));
    }
}

task.hpp

typedef void (*mission)() ;
std::vector<mission> missionarray;
void task1()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task1"<<std::endl;
}
void task2()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task2"<<std::endl;
}
void task3()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task3"<<std::endl;
}
void task4()
{
    std::cout<<"I am childprocess: "<<getpid()<<" running task4"<<std::endl;
}

运行截图：

结语

那么这就是本期博客关于进程池的详细介绍了，那么从进程池的意义以及进程池的实现大体框架到具体细节这几个维度带你全面解析进程池，其次注意就是进程池的应用场景一定是要执行任务数量要大于子进程的数量，如果你要执行30个任务，创建27个子进程其实意义不大，那么读者下来也可以自己实现一个属于你自己的进程池，那么我的下一期博客会介绍命名管道，那么我会持续更新，希望您能够多多关注哦，如果本篇文章有帮组到你，还请三连加关注哦，你的支持就是我创作的最大动力！