如何实现进程池

时间：2021-07-08 21:50:01
如何实现进程池


http://blog.csdn.net/guosha/article/details/3874998 这是原作者链接 感谢作者分享

有位站友问，如何在linux下实现进程池技术，原贴见：
http://topic.csdn.net/u/20090206/16/b424e1c1-90dc-4589-a63f-1d90ed6560ae.html。

之前虽对进程池这个名词早有耳闻，但一直没有真正接触过。乍一听好像有点复杂，但稍微一想却也简单，下面我就按自己的想法来实现一个简单的模型。

 

跟最简单的资源管理一样，进程池技术的应该致少由以下两部分组成

资源进程
预先创建好的空闲进程，管理进程会把工作分发到空闲进程来处理。

管理进程
管理进程负责创建资源进程，把工作交给空闲资源进程处理，回收已经处理完工作的资源进程。

 

上面资源进程跟管理进程的概念很好理解，下面就是进程池的关键，管理进程如何有效的管理资源进程，如何分配任务给资源进程，如何回收空闲资源进程。其实稍加分析一下也很简单，管理进程要有效的管理资源进程，那么管理进程跟资源进程间必然需要交互，如何交互呢？linux下两个进程如何交互？IPC呗，信号，信号量，消息队列，管道，随便选一个合适的就可以了。


下面我就用信号加消息队列来完成上面帖子里的进程池需求。

[cpp] view plain copy print ?
  1. //主接收网络消息进程  
  2. int main()  
  3. {  
  4.     int iReLen;  
  5.     char szBuffer[64];  
  6.     ...  
  7.     //初始化进程池  
  8.     InitProcessPoll();  
  9.     ....  
  10.     while(1)  
  11.     {  
  12.         iRelen = recv(iSock, szBuffer, sizeof(szbuffer), 0);  
  13.         if (iRelen > 0)  
  14.         {  
  15.             SubmitWork(szBuffer, iRelen);   //提交工作到资源进程  
  16.         }     
  17.     }  
  18.     ...  
  19. }  
  20.   
  21. //  
  22. int SubmitWork(void *pBuffer, int iMsgLen)  
  23. {  
  24.     int iMsgQueID;  
  25.       
  26.     iMsgQueID = GetMsgQue(key, 0); //取得跟管理进程通信的消息队更句柄;  
  27.       
  28.     msgsnd(iMsgQueID, pBuffer, iMsgLen, 0);  
  29. }  
  30.   
  31. int InitProcessPoll(const int iProcessNum)  
  32. {  
  33.     int iPid;  
  34.       
  35.     //创建管理进程  
  36.     iPid = fork();  
  37.     if (iPid == 0)  
  38.     {  
  39.         InitMngProcess(iProcessNum);  
  40.     }     
  41.     return 0;  
  42. }  
  43.   
  44. typedef struct  
  45. {  
  46.     int pid;  
  47.     int iFlag;  
  48. } T_ProcessStatus;  
  49.   
  50. //指向资源进程管理结构  
  51. T_ProcessStatus *pProcessMng = NULL；  
  52.   
  53. //记录有总共有多少个资源进程  
  54. INT32 iMaxProcessNum = 0;  
  55.   
  56. //初始管理进程管理结构并创建资源子进程，最后接收外部工作请求并分发到资源进行进行处理  
  57. InitMngProcess(const int iProcessNum)  
  58. {  
  59.     int i;  
  60.     int iPid;  
  61.     int iRtn;  
  62.     int iMsgLen;  
  63.     int iMsgQue1, iMsgQue2;  
  64.     char szBuffer[64];  
  65.       
  66.     //创建管理进程结构  
  67.     pProcessMng = calloc(iProcessNum, sizeof(T_ProcessStatus))  
  68.   
  69.     for (i = 0; i < iProcessNum; i++)  
  70.     {  
  71.         iPid = fork();  
  72.         if (iPid == 0);  
  73.         {  
  74.             //资源进程;  
  75.             ResourceProcess();  
  76.         }  
  77.         pProcessMng[i].pid = iPid; //记录资源进程的进程号  
  78.         pProcessMng[i].iFlag = 0;   //把资源进程置为空闲  
  79.     }  
  80.       
  81.     iMaxProcessNum = iProcessNum;  
  82.   
  83.     //创建外部跟管理进程通信的消息队列；  
  84.     iMsgQue1 = CreateMsgQue();  
  85.   
  86.     //创建管理进程跟资源进程通信的消息队列；  
  87.     iMsgQue2 = CreateMsgQue();  
  88.   
  89.     //安装资源进程回收信号处理函数  
  90.     signal(SIGUSR1, ReleaseAProcess);     
  91.   
  92.     //开始接收外部传入的任务  
  93.     while(1)  
  94.     {  
  95.         //接收外部的工作请求  
  96.         iMsgLen = msgrcv(iMsgQue1, szBuffer, sizeof(szBuffer), 0);  
  97.           
  98.         //转发工作请求到资源进程处理消息队列  
  99.         iRtn = msgsnd(iMsgQue2, szBuffer, iMsgLen, 0);  
  100.       
  101.         //通知其中的一个空闲资源进程进行处理  
  102.         NoticeAIdleProcess();  
  103.     }  
  104. }  
  105.   
  106. //通知一个空闲资源进程进行处理  
  107. int NoticeAIdleProcess()  
  108. {  
  109.     int i;  
  110.       
  111.     for (i = 0; i < iMaxProcessNum; i++)  
  112.     {  
  113.         if (pProcessMng[i].iFlag == 0)  
  114.         {  
  115.             pProessMng[i].Flag = 1;  
  116.             kill(processMng[i].pid, SIGUSR1);  
  117.             return 0;  
  118.         }  
  119.     }  
  120.       
  121.     return -1;    
  122. }  
  123.   
  124. //回收一个处理完成的资源进程  
  125. void ReleaseAProcess(int iPid)  
  126. {  
  127.     int i;  
  128.       
  129.     for (i = 0; i < iMaxProcessNum; i++)  
  130.     {  
  131.         if (pProessMng[i].pid == iPid)  
  132.         {  
  133.             pProcessMng[i].iFlag = 0;  
  134.             return;  
  135.         }  
  136.     }  
  137.   
  138.     return;  
  139. }  
  140.   
  141. //资源进程的处理  
  142. void ResourceProcess()  
  143. {  
  144.     //安装有工作通知信号处理  
  145.     signal(SIGUSR1, SIG_IGN);  
  146.   
  147.     //设置只对SIGUSR1信号感兴趣  
  148.     sigprocmask()  
  149.     while(1)  
  150.     {  
  151.         //没有消息处理时挂起  
  152.         pause();  
  153.           
  154.         //处理工作  
  155.         void StartWork()  
  156.   
  157.         //通知管理进程工作处理完成  
  158.         NoticeMngProcessFinishedWork();   
  159.     }  
  160. }  
  161.   
  162. //处理消息  
  163. void StartWork()  
  164. {  
  165.     char szBuffer[64];  
  166.     int iMsgID;  
  167.     int iRtn;  
  168.       
  169.     iMsgID = msgget();  
  170.     iRtn = msgrcv(iMsgID, szBuffer, sizeof(szBuffer), 0);  
  171.       
  172.     //现在在子进程里取得了需要处理的消息，可以开始处理该消息直到消息处理完成  
  173.   
  174.     return;   
  175. }  
  176.   
  177. //通知管理进程回收资源进程  
  178. void NoticeMngProcessFinishedWork();  
  179. {  
  180.     kill(MngProcessPid, SIGUSR1);  
  181.       
  182.     return;  
  183. }  

 

如上，一个基本的进程池处理模型就建立好了, 当然你不能直接拿到编译，因为里面很多都是伪代码，另个上面的模型忽略了很多细节处理，很多地方是很不可靠的。

现在你是不是对进程池技术更了解了呢？


标签：

相关文章