Mehrere Prozesse mit PHP starten, um Daten zu verarbeiten.

Question

Ich habe eine Warteschlange (Amazon SQS) von Daten, die verarbeitet werden müssen, und ich möchte es mit mehreren Prozessen (in PHP) zu tun.

Ich möchte, dass die untergeordneten Arbeiter so etwas tun (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);
}

Es muss immer 1 Kindprozess laufen, aber in Notzeiten möchte ich einen Kindprozess (forken?), damit er die Warteschlange schneller abarbeiten kann.

Kann mir jemand mit einem groben PHP-Skelett von dem, was ich brauche, helfen, oder mich in die richtige Richtung weisen?

Ich glaube, ich muss einen Blick auf http://php.net/manual/en/function.pcntl-fork.php aber ich bin mir nicht sicher, wie ich damit mehrere Prozesse verwalten kann.

Answer 1

Wenn Sie einen Prozess aufspalten, erstellen Sie ein Duplikat dieses Prozesses. Mit anderen Worten: Die Kopie (Fork) enthält alles, was der ursprüngliche Prozess hatte (einschließlich der Datei-Handles)

Woher wissen Sie also, ob Sie der übergeordnete oder der abgespaltene Prozess sind?

Das Beispiel auf der verlinkten Seite zeigt dies recht deutlich

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

?>

Um dies auf das zu erweitern, was Sie wollen

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

?>

Dies wird einen Fork-Prozess erzeugen (in diesem Fall eine Verschwendung) und eine Schleife verwenden, um mehrere Prozesse zu erzeugen. Wenn der Kindprozess fertig ist, wird exit den Kindprozess beenden und pcntl_wait() wird zurückkehren und dem Elternprozess erlauben, weiterzumachen. Ich bin mir bei php nicht sicher, aber wenn der Elternprozess stirbt oder beendet wird, wird der Kindprozess auch dann beendet, wenn das Kind noch nicht fertig ist, daher pcntl_wait.

Vielleicht sollten Sie sich statt des Forkings lieber die Palette der exec-Funktionen ansehen?

Ein Vorbehalt.

Der Forking-Prozess kann mit Problemen behaftet sein: Datenbank-Handles werden geschlossen, wenn ein Child beendet wird usw. Sie können auch einen Server mit zu vielen Prozessen zu töten, wenn etwas schief geht. verbringen viel Zeit spielen und testen und lesen.

DC

Answer 2

Ich weiß, dass dies ein alter Thread ist, aber es sah so aus, als könnte er eine vollständigere Antwort gebrauchen. So erzeuge ich im Allgemeinen mehrere Prozesse in PHP.

Ein Wort der Warnung: PHP war zum Sterben bestimmt. Das bedeutet, dass die Sprache einige Sekunden lang ausgeführt und dann beendet werden sollte. Obwohl die Speicherbereinigung in PHP einen langen Weg zurückgelegt hat, sollten Sie vorsichtig sein. Überwachen Sie Ihre Prozesse auf unerwarteten Speicherverbrauch oder andere Seltsamkeiten. Beobachten Sie alles eine Zeit lang wie ein Falke, bevor Sie es einstellen und vergessen, und selbst dann sollten Sie die Prozesse ab und zu überprüfen oder sie automatisch benachrichtigen lassen, wenn etwas nicht stimmt.

Als ich das hier abtippte, schien es mir eine gute Idee zu sein, das hier anzubringen github auch.

Wenn Sie bereit sind, das Programm auszuführen, empfehle ich Ihnen, das Protokoll mit tail -f abzufragen, um die Ausgabe zu sehen.

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