用PHP产生多个进程来处理数据。

Question

我有一个需要处理的数据队列（Amazon SQS），我想用多个进程（在PHP中）来完成。

我希望孩子工人做这样的事情（pseduoish代码）：

 


while(true) { 

    $array = $queue->fetchNItems(10); // get 10 items 

    if(!count($array)) 
     killProcess(); 

    foreach($array as $item) { 
     ... // process the item 
     $queue->remove($item); 
    } 

    sleep(2); 
} 

 

我总是需要1个进程正在运行，但在需要的时候我想（叉？）孩子进程，以便它可以帮助更快地处理队列。

有人可以帮助我一个粗糙的PHP骨架，我需要什么，或指向我在正确的方向吗？

我想我需要看看http://php.net/manual/en/function.pcntl-fork.php，但我不知道如何使用它来管理多个进程。

Answer 1

当你fork一个进程。你重复这个过程。换句话说，副本（fork）包含原始进程的所有内容（包括文件句柄）

那么，如何知道您是父进程还是分叉进程？

从链接页的例子显示了这个漂亮的明确

<?php 

$pid = pcntl_fork(); 
if ($pid == -1) { 
    die('could not fork'); 
} else if ($pid) { 
    // we are the parent 
    pcntl_wait($status); //Protect against Zombie children 
} else { 
    // we are the child 
} 

?> 

将其扩展到您想要

<?php 

$pid = pcntl_fork(); 
if ($pid == -1) { 
    die('could not fork'); 
} else if ($pid) { 
    // we are the parent 
    pcntl_wait($status); //Protect against Zombie children 
} else { 
    // we are the child 
    while(true) { 

     $array = $queue->fetchNItems(10); // get 10 items 

     if(!count($array)) { 
      exit(); 
     } 

     foreach($array as $item) { 
       ... // process the item 
       $queue->remove($item); 
     } 

     sleep(2); 
    } 
} 

?> 

这将在派生进程创造什么（在这种情况下浪费）使用循环来创建多个进程。当孩子完成退出时会杀死孩子的过程。和pcntl_wait（）将返回，允许父母继续。我不确定关于PHP，但如果父进程死亡或退出，即使孩子没有完成，它也会终止子进程。因此pcntl_wait。如果您产生多个孩子，则需要更复杂的系统。

也许而不是分叉你应该看一下exec函数的范围？

一个告诫。

分叉过程可能会产生问题，当孩子退出时数据库句柄被关闭等。如果出现问题，您也可以使用多个进程来终止服务器。花费大量的时间玩和测试和阅读。

DC

Answer 2

我知道这是一个古老的线程，但看起来像它可以使用一个更完整的答案。这是我通常在PHP中产生多个进程的方式。

谨慎的一句话：PHP was meant to die.意思是说，语言意味着要执行几秒钟然后退出。尽管PHP中的垃圾清理已经走过了很长的路，但要小心。监视您的进程，以发现意外的内存消耗或其他异常情况。在你设置并忘记它之前，一切都像鹰一样观看，即使如此，仍然会偶尔检查一下流程，或者在出现问题时自动通知它们。

当我打字的时候，看起来像是一个好主意，拍到github。

当准备运行该程序时，我建议在日志上做一个尾部-f来查看输出。

<?php 
/* 
* date: 27-sep-2015 
* auth: robert smith 
* info: run a php daemon process 
* lic : MIT License (see LICENSE.txt for details) 
*/  
$pwd = realpath(""); 

$daemon = array(
    "log"  => $pwd."/service.log", 
    "errorLog" => $pwd."/service.error.log", 
    "pid_file" => $pwd."/", 
    "pid"  => "", 
    "stdout" => NULL, 
    "stderr" => NULL, 
    "callback" => array("myProcessA", "myProcessB") 
); 

/* 
* main (spawn new process) 
*/ 
foreach ($daemon["callback"] as $k => &$v) 
    { 
    $pid = pcntl_fork(); 

    if ($pid < 0) 
    exit("fork failed: unable to fork\n"); 

    if ($pid == 0) 
    spawnChores($daemon, $v); 
    } 

exit("fork succeeded, spawning process\n"); 
/* 
* end main 
*/ 

/* 
* functions 
*/ 
function spawnChores(&$daemon, &$callback) 
    { 
    // become own session 
    $sid = posix_setsid(); 

    if ($sid < 0) 
    exit("fork failed: unable to become a session leader\n"); 

    // set working directory as root (so files & dirs are not locked because of process) 
    chdir("/"); 

    // close open parent file descriptors system STDIN, STDOUT, STDERR 
    fclose(STDIN); 
    fclose(STDOUT); 
    fclose(STDERR); 

    // setup custom file descriptors 
    $daemon["stdout"] = fopen($daemon["log"], "ab"); 
    $daemon["stderr"] = fopen($daemon["errorLog"], "ab"); 

    // publish pid 
    $daemon["pid"] = sprintf("%d", getmypid()); 
    file_put_contents($daemon["pid_file"].$callback.".pid", $daemon["pid"]."\n"); 

    // publish start message to log 
    fprintf($daemon["stdout"], "%s daemon %s started with pid %s\n", date("Y-M-d H:i:s"), $callback, $daemon["pid"]); 

    call_user_func($callback, $daemon); 

    // publish finish message to log 
    fprintf($daemon["stdout"], "%s daemon %s terminated with pid %s\n", date("Y-M-d H:i:s"), $callback, $daemon["pid"]); 

    exit(0); 
    } 

function myProcessA(&$daemon) 
    { 
    $run_for_seconds = 30; 
    for($i=0; $i<$run_for_seconds; $i++) 
    { 
    fprintf($daemon["stdout"], "Just being a process, %s, for %d more seconds\n", __FUNCTION__, $run_for_seconds - $i); 
    sleep(1); 
    } 
    } 

function myProcessB(&$daemon) 
    { 
    $run_for_seconds = 30; 
    for($i=0; $i<$run_for_seconds; $i++) 
    { 
    fprintf($daemon["stdout"], "Just being a process, %s, for %d/%d seconds\n", __FUNCTION__, $i, $run_for_seconds); 
    sleep(1); 
    } 
    } 
?>