PHP多进程处理并行处理任务实例

2019独角兽企业重金招聘Python工程师标准>>>  
本文目的
本文通过例子讲解linux环境下，使用php进行并发任务处理，以及如何通过pipe用于进程间的数据同步。写得比较简单，作为备忘录。
PHP多进程
通过pcntl_XXX系列函数使用多进程功能。注意：pcntl_XXX只能运行在php CLI（命令行）环境下，在web服务器环境下，会出现无法预期的结果，请慎用！
管道PIPE
管道用于承载简称之间的通讯数据。为了方便理解，可以将管道比作文件，进程A将数据写到管道P中，然后进程B从管道P中读取数据。php提供的管道操作API与操作文件的API基本一样，除了创建管道使用posix_mkfifo函数，读写等操作均与文件操作函数相同。当然，你可以直接使用文件模拟管道，但是那样无法使用管道的特性了。
僵尸进程
子进程结束时，父进程没有等待它（通过调用wait或者waitpid）,那么子进程结束后不会释放所有资源（浪费呀！），这种进程被称为僵尸进程，他里面存放了子进程结束时的相关数据，如果僵尸进程过多，会占用大量系统资源（如内存），影响机器性能。
代码
废话少说直接上代码
 /** 
   * this is a demo for php fork and pipe usage. fork use 
   * to create child process and pipe is used to sychoroize 
   * the child process and its main process. 
   * @author bourneli 
   * @date: 2012-7-6 
   */ 
 define( "PC" , 10);  // 进程个数 
 define( "TO" , 4);  // 超时 
 define( "TS" , 4);  // 事件跨度，用于模拟任务延时 
 if  (!function_exists( 'pcntl_fork' )) { 
      die ( "pcntl_fork not existing" ); 
 } 
 // 创建管道 
 $sPipePath  =  "my_pipe." .posix_getpid(); 
 if  (!posix_mkfifo( $sPipePath , 0666)) { 
      die ( "create pipe {$sPipePath} error" ); 
 } 
 // 模拟任务并发 
 for  ( $i  = 0;  $i  < PC; ++ $i  ) { 
      $nPID  = pcntl_fork();  // 创建子进程 
      if  ( $nPID  == 0) { 
          // 子进程过程 
          sleep(rand(1,TS));  // 模拟延时 
          $oW  =  fopen ( $sPipePath ,  'w' ); 
          fwrite( $oW ,  $i . "\n" );  // 当前任务处理完比，在管道中写入数据 
          fclose( $oW ); 
          exit (0);  // 执行完后退出 
      } 
 } 
 // 父进程 
 $oR  =  fopen ( $sPipePath ,  'r' ); 
 stream_set_blocking( $oR , FALSE);  // 将管道设置为非堵塞，用于适应超时机制 
 $sData  =  '' ;  // 存放管道中的数据 
 $nLine  = 0; 
 $nStart  = time(); 
 while  ( $nLine  < PC && (time() -  $nStart ) < TO) { 
      $sLine  =  fread ( $oR , 1024); 
      if  ( empty ( $sLine )) { 
          continue ;    
      }    
       
      echo  "current line: {$sLine}\n" ; 
      // 用于分析多少任务处理完毕，通过‘\n’标识 
      foreach ( str_split ( $sLine )  as  $c ) { 
          if  ( "\n"  ==  $c ) { 
              ++ $nLine ; 
          } 
      } 
      $sData  .=  $sLine ; 
 } 
 echo  "Final line count:$nLine\n" ; 
 fclose( $oR ); 
 unlink( $sPipePath );  // 删除管道，已经没有作用了 
 // 等待子进程执行完毕，避免僵尸进程 
 $n  = 0; 
 while  ( $n  < PC) { 
      $nStatus  = -1; 
      $nPID  = pcntl_wait( $nStatus , WNOHANG); 
      if  ( $nPID  > 0) { 
          echo  "{$nPID} exit\n" ; 
          ++ $n ; 
      } 
 } 
 // 验证结果，主要查看结果中是否每个任务都完成了 
 $arr2  =  array (); 
 foreach ( explode ( "\n" ,  $sData )  as  $i ) { // trim all 
      if  ( is_numeric (trim( $i ))) { 
          array_push ( $arr2 ,  $i );   
      } 
 } 
 $arr2  =  array_unique ( $arr2 ); 
 if  (  count ( $arr2 ) == PC) {   
      echo  'ok' ;  
 }  else  { 
      echo   "error count "  .  count ( $arr2 ) .  "\n" ; 
      var_dump( $arr2 ); 
 } 


ok，完毕，注释写的比较清除，执行结果如下：

  