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

发布于 2024-08-16 21:28:57 字数 638 浏览 3 评论 0原文

我有一个需要处理的数据队列 (Amazon SQS)，并且我想通过多个进程（在 PHP 中）来完成此操作。

我希望子进程执行类似这样的操作（伪代码）：



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，但我不确定如何使用它来管理多个进程。

原文

I have a queue (Amazon SQS) of data that needs to be processed, and I would like to do it with multiple processes (in PHP).

I want the child workers to do something like this (pseduoish code):



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);
}

I always need 1 child process to be running, but in times of need I want to (fork?) a child process so that it can help process the queue faster.

Can someone help me with a rough PHP skeleton of what I need, or point me in the right direction?

I think I need to take a look at http://php.net/manual/en/function.pcntl-fork.php, but I'm not sure how I can use this to manage multiple processes.

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

凑诗 2024-08-23 21:28:57

当你 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 函数的范围而不是分叉？

一个警告。

分叉进程可能会出现问题，当子进程退出时数据库句柄被关闭等。如果出现问题，您还可以杀死具有多个进程的服务器。花很多时间去玩、测试和阅读。

直流

When you fork a process. you make a duplicate of that process. In other words the copy (fork) contains everything the original process had (including file handles)

So how do you know if you are the parent or the forked process?

The example from the linked page shows this pretty clearly

<?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
}

?>

To extend this to what you want

<?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);
     }
}

?>

This will create on forked process ( a waste in this instance ) use a loop to create multiple processes. when the child has finished exit will kill the child process. and pcntl_wait() will return allowing the parent to continue. I am not sure about php but if the parent process dies or exits it will kill the child process even if the child is not finished. hence the pcntl_wait. a more elaborate system is required if you spawn multiple children.

perhaps rather than forking you should look at the range of exec functions?

A caveat.

forking process can be wrought with problems, database handles being closed when a child exits etc. You can also kill a server with to many processes if something goes wrong. spend a lot of time playing and testing and reading.

回复收藏 0 原文

甜柠檬 2024-08-23 21:28:57

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

需要注意的是：PHP 注定要消亡。意思是，该语言意味着执行几秒钟然后退出。尽管 PHP 中的垃圾清理已经取得了长足的进步，但要小心。监视您的进程是否存在意外的内存消耗或其他异常情况。在设置之前像鹰一样观察一切，然后忘记它，即便如此，仍然偶尔检查一下进程，或者让它们在出现问题时自动通知。

当我输入此内容时，将其放在 github 上似乎也是个好主意。

当准备好运行程序时，我建议在日志上执行 tail -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);
    }
  }
?>

I know this is an old thread, but looked like it could use a more complete answer. This is how I generally spawn multiple processes in PHP.

A word of caution: PHP was meant to die. Meaning, the language was mean to execute for a few seconds then exit. Though, garbage cleanup in PHP has come a long way, be careful. Monitor your processes for unexpected memory consumption, or other oddities. Watch everything like a hawk for a while before you set it and forget it, and even then, still check the processes once in a while or have them automatically notify if something becomes amiss.

As I was typing this up, seemed like a good idea to slap it on github too.

When ready to run the program, I recommend, doing a tail -f on the log to see the output.

<?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);
    }
  }
?>

回复收藏 0 原文

~没有更多了~