多線程編程-通過fork方式創建任務進程

2019-11-06 06:05:38

字體：大中小

來源：轉載

供稿：網友

背景

在多進程中需要創建一個任務進程進行待處理任務的分發工作。也就是說，需要在主進程中detach剝離出一個進程，且該子進程獨立于主進程，進行獨立的任務發送。如果是單純在主進程中fork一個子進程的話，如果進行了wait等待操作，那么該任務進程就失去了獨立性，無法實現邊取任務邊分發任務。而如果是不進行wait等待操作的話，那么子進程可能也會執行后續創建多個處理任務的進程，使得出現的并行進程數量為預期的double。當然這也是可以通過在并發多進程wait之前對任務分配進程進行wait操作來達到預期要求。本文采取兩種方式來實現任務分配進程和任務處理進程的共存。

方案1：

雙fork方式，以孫進程作為detach主進程的任務分配進程，由于該進程是游離于主進程的，所以其回收是通過系統實現的。

代碼

void childPRocnew_hash_test(){ struct msgstrunew_hash_align msgs; int msgid,ret_value; char str[512]; /* First, we set up the message queue. */ // msgid = msgget((key_t)MSGKEY, 0666 | ipC_CREAT);//該鍵值則唯一對應一個消息隊列 while(1) { msgid = msgget(MSGKEY,IPC_EXCL );/*檢查消息隊列是否存在 */ if(msgid < 0){ printf("msq not existed! errno=%d [%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } /*接收消息隊列*/ ret_value = msgrcv(msgid,&msgs,sizeof(struct msgstrunew_hash_align),0,0); std::cout<<"get task pid="<<getpid()<<","<<msgs.msgtext<<std::endl; usleep(100); if(ret_value == -1) { fprintf(stderr, "msgrcv failed with error: %d/n", errno);//消息隊列中的信息被取完？？ exit(EXIT_FAILURE);//消息隊列為空的時候，就跳出。也可以設計成，消息隊列為空時，不跳出，而是等待。 } else { std::string hashline = msgs.msgtext;//接收到的是一行的信息,需要將該行的數據進行切分 // std::cout<<hashline.c_str()<<std::endl;// if (hashline.size()<10) { printf("datasize=%d,data=%s/n",hashline.size(),hashline.c_str()); exit(EXIT_SUCCESS);//換成break的效果呢？？？是不一樣的啊 } } //因為在send的時候，只send了一個end，當該標志信息被讀取之后，其他的進程自然是讀取不到信息的， } return;}void SendHashLine(){ struct msgstrunew_hash_align msgs; int msg_type; char str[80]; int ret_value; int msqid; msqid=msgget(MSGKEY,IPC_EXCL ); /*檢查消息隊列是否存在*/ if(msqid < 0){ msqid = msgget(MSGKEY,IPC_CREAT|0666);/*創建消息隊列*/ if(msqid <0){ printf("failed to create msq | errno=%d [%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } } ifstream m_InStream("hash_list.txt"); string oneline; int running=1; int linenun=0; while (running) { if(!getline(m_InStream, oneline,'/n')) { running = 0; for(int i = 0; i<ChildNum;++i) { msgs.msgtype = linenun; snprintf(msgs.msgtext,sizeof(msgs.msgtext),"%s","nil");//strncpy(response.m_ClientIp, request.m_ClientIp, sizeof(response.m_ClientIp));//記錄客戶端的ip,昵稱服務需要 // msgs.msgtext = "nil"; // strcpy(msgs.msgtext, str); /* 發送消息隊列 */ std::cout<<"task deliver pid="<<getpid()<<","<<msgs.msgtext<<std::endl; ret_value = msgsnd(msqid,&msgs,sizeof(struct msgstrunew_hash_align),0);//消息隊列標識符，準備發現信息的指針，信息的長度，控制標志位 if ( ret_value < 0 ) { printf("msgsnd() write msg failed,errno=%d[%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } } break; } linenun++; strncpy(str,oneline.c_str(),80); msgs.msgtype = linenun; // msgs.msgtext = str; strncpy(msgs.msgtext,str,sizeof(msgs.msgtext)); // strcpy(msgs.msgtext, str); /* 發送消息隊列 */ std::cout<<"task deliver pid="<<getpid()<<","<<msgs.msgtext<<std::endl; int test_len = sizeof(struct msgstrunew_hash_align); ret_value = msgsnd(msqid,&msgs,sizeof(struct msgstrunew_hash_align),0);//消息隊列標識符，準備發現信息的指針，信息的長度，控制標志位 // sleep(1); if ( ret_value < 0 ) { printf("msgsnd() write msg failed,errno=%d[%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } }}void MultiOldAligndata_test(){ //先用一個進程進行任務的創建,即用一個進程做消息隊列的創建 pid_t fpid; //fpid表示fork函數返回的值 int count=0; fpid=fork(); if (fpid < 0) printf("error in fork!"); else if (fpid == 0) { printf("i am the child process, my process id is %d/n",getpid()); printf("子進程/n"); count++; pid_t fpid_2; fpid_2 = fork(); if(fpid_2 <0) { printf("error in fork_second!/n"); } else if(fpid_2 == 0) { //該孫進程處理自己想要做的事情，即做任務的分配 SendHashLine();// } exit(EXIT_SUCCESS); } else { printf("i am the parent process, my process id is %d/n",getpid()); count++; } int status_1 =0; int mpid_1 =0; mpid_1 = wait(&status_1); //基于消息隊列的多進程版本----------begin---------------------------- time_t start,stop; start = time(NULL); int data_processed; string some_data; int i,cpid; // /* create 5 child process */ for (i=0;i<ChildNum;i++){ cpid = fork(); if (cpid < 0) printf("fork failed/n"); else if (cpid ==0) /*child process*/ { childprocnew_hash_test(); } } int status =0; int mpid =0; for(int i=0;i<ChildNum;i++) { mpid = wait(&status); printf("pid[%d] is exit with status[%d]/n",mpid,status); } stop = time(NULL); std::cout<<"cost:"<<double(stop-start)<<std::endl; //基于消息隊列的多進程版本----------end----------------------------

如果采用主進程直接fork一個子進程作為任務分配進程的話，該任務分配子進程也會執行后期的任務處理進程的多進程創建，使得并行的進程數是預期的double。且由于主進程的等待操作并沒有考慮到這多出來的一倍進程，導致這部分進程沒有退出，僵死狀態。同時，任務分配進程也沒有退出。所以此時通過ps命令查看，發現進程數為7（=5并行進程+2（任務分配進程和主進程））并行執行過程輸出界面信息：這里寫圖片描述 ps查看進程情況：可以查看對應的進程號是與上面多出來的一倍進程相對應的。

方案2：

通過主進程fork一個子進程作為任務分配進程。主進程創建多個并發的任務處理進程。且在主進程wait各個任務處理進程之前進行一次任務進程的wait操作，使得任務進程得以回收，再進程并發進程的wait操作。

代碼：

//exit with statusvoid childprocnew_hash_test(){ struct msgstrunew_hash_align msgs; int msgid,ret_value; char str[512]; /* First, we set up the message queue. */ // msgid = msgget((key_t)MSGKEY, 0666 | IPC_CREAT);//該鍵值則唯一對應一個消息隊列 while(1) { msgid = msgget(MSGKEY,IPC_EXCL );/*檢查消息隊列是否存在 */ if(msgid < 0){ printf("msq not existed! errno=%d [%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } /*接收消息隊列*/ ret_value = msgrcv(msgid,&msgs,sizeof(struct msgstrunew_hash_align),0,0); std::cout<<"get task pid="<<getpid()<<","<<msgs.msgtext<<std::endl; usleep(100); if(ret_value == -1) { fprintf(stderr, "msgrcv failed with error: %d/n", errno);//消息隊列中的信息被取完？？ exit(EXIT_FAILURE);//消息隊列為空的時候，就跳出。也可以設計成，消息隊列為空時，不跳出，而是等待。 } else { std::string hashline = msgs.msgtext;//接收到的是一行的信息,需要將該行的數據進行切分 if (hashline.size()<10) { printf("datasize=%d,data=%s/n",hashline.size(),hashline.c_str()); exit(EXIT_SUCCESS);//換成break的效果呢？？？是不一樣的啊 } } } return;}void SendHashLine(){ struct msgstrunew_hash_align msgs; int msg_type; char str[80]; int ret_value; int msqid; msqid=msgget(MSGKEY,IPC_EXCL ); /*檢查消息隊列是否存在*/ if(msqid < 0){ msqid = msgget(MSGKEY,IPC_CREAT|0666);/*創建消息隊列*/ if(msqid <0){ printf("failed to create msq | errno=%d [%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } } ifstream m_InStream("hash_list.txt"); string oneline; int running=1; int linenun=0; while (running) { if(!getline(m_InStream, oneline,'/n')) { running = 0; for(int i = 0; i<ChildNum;++i) { msgs.msgtype = linenun; snprintf(msgs.msgtext,sizeof(msgs.msgtext),"%s","nil"); /* 發送消息隊列 */ std::cout<<"task deliver pid="<<getpid()<<","<<msgs.msgtext<<std::endl; ret_value = msgsnd(msqid,&msgs,sizeof(struct msgstrunew_hash_align),0); if ( ret_value < 0 ) { printf("msgsnd() write msg failed,errno=%d[%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } } break;//可以對比下述兩種退出方式 //exit(EXIT_SUCCESS);//EXIT_SUCCESS } linenun++; SplitString(oneline.c_str(),oneline.size(),"/t",scidlist); strncpy(str,oneline.c_str(),80); msgs.msgtype = linenun; strncpy(msgs.msgtext,str,sizeof(msgs.msgtext)); /* 發送消息隊列 */ std::cout<<"task deliver pid="<<getpid()<<","<<msgs.msgtext<<std::endl; int test_len = sizeof(struct msgstrunew_hash_align); ret_value = msgsnd(msqid,&msgs,sizeof(struct msgstrunew_hash_align),0); if ( ret_value < 0 ) { printf("msgsnd() write msg failed,errno=%d[%s]/n",errno,strerror(errno)); exit(EXIT_FAILURE); } }}void MultiOldAligndata_test(){ pid_t fpid; int count=0; fpid=fork(); if (fpid < 0) printf("error in fork!"); else if (fpid == 0) { printf("i am the child process, my process id is %d/n",getpid()); printf("子進程/n"); SendHashLine();//任務分配進程 } else { printf("i am the parent process, my process id is %d/n",getpid()); count++; //基于消息隊列的多進程版本----------begin---------------------------- time_t start,stop; start = time(NULL); int data_processed; string some_data; int i,cpid; // /* create 5 child process */ for (i=0;i<ChildNum;i++){ cpid = fork(); if (cpid < 0) printf("fork failed/n"); else if (cpid ==0) /*child process*/ { childprocnew_hash_test(); } } //需要等待任務分配的進程 int status_1 =0; int mpid_1 =0; mpid_1 = wait(&status_1); int status =0; int mpid =0; for(int i=0;i<ChildNum;i++) { mpid = wait(&status); printf("pid[%d] is exit with status[%d]/n",mpid,status); } stop = time(NULL); std::cout<<"cost:"<<double(stop-start)<<std::endl; //基于消息隊列的多進程版本----------end---------------------------- } // int status_1 =0; // int mpid_1 =0; // mpid_1 = wait(&status_1);//不等待的話？？？ // sleep(10);//不加sleep操作的話則任務進程會先退出？？？因為任務進程最先執行結束}

上一篇：tomcat配置編碼格式

下一篇：GCC -D選項