RabbitMQ-Consumer mit PHP starten

Message Queues wie RabbitMQ sind eine schöne und einfache Möglichkeit, PHP-Prozesse in den Hintergrund zu verlagern. Anstatt eine zeitaufwendige Aufgabe innerhalb eines Web-Requests auszuführen, schiebt man einfach eine Nachricht in eine Queue und holt sich das Ergebnis später ab. Klingt im ersten Schritt einfach. Aber wenn es darum geht, PHP-Prozesse dauerhaft im Hintergrund laufen zu lassen, um diese Nachrichten aus einer Queue zu konsumieren, kann es unangenehm werden.

Auf einer dedizierten Linux-Maschine ist das Leben leicht. Aber was tun Sie, wenn das nicht der Fall ist? Unter Linux kann man einfach Werkzeuge wie supervisord einsetzen, um das zu steuern. Aber was machen Sie, wenn Sie auf Shared Hosting laufen müssen (der Rabbit muss dann natürlich irgendwo anders verfügbar sein) und keine zusätzliche Software installieren können? Noch schwieriger: Was tun Sie unter Windows? Ich würde niemals empfehlen, einen Supervisor-Daemon auf einer Windows-Maschine zu betreiben, auch wenn andere das tun. Der Linux-Cron-Daemon (meist auch auf Shared Hosting verfügbar) oder die geplanten Aufgaben von Windows können eine Lösung sein. Schreiben Sie ein kleines PHP-Skript, das alle paar Minuten ausgeführt wird, sich mit RabbitMQ verbindet und auf Nachrichten wartet.

(Die Beispiele basieren auf php-amqplib)

// connect to your RabbitMQ server
$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// declare a channel
$channel->queue_declare('myQueue', false, false, false, false);

// define consuming
$channel->basic_consume('hello', '', false, true, false, false,
  function($msg) {
    // a callback to consume incoming messages
    echo "Message received:n";
    print_r($msg->body);
  }
);

// wait for incoming messages
while(count($channel->callbacks)) {
  $channel->wait();
}

Dieses Snippet zeigt, wie einfach es sein kann, Nachrichten aus einer Queue zu konsumieren. Wenn Sie solchen Code in ein PHP-Skript packen und das Skript jede Minute über den Linux-Cron-Daemon oder eine geplante Windows-Aufgabe ausführen, würde jede Minute ein neuer Consumer erzeugt. Das ist nicht ganz das, was wir wollen. Wie steuere ich nun die Anzahl der gestarteten Consumer?

Dazu müssen wir lediglich die Rückgabewerte von queue_declare() prüfen. Einer der Rückgabewerte ist die aktuelle Anzahl der Consumer, die auf der Queue lauschen.

// It return an array with different values.
// The third one is the consumer count.
list(,,$consumerCount) = $channel->queue_declare('myQueue', false, false, false, false);

if ($consumerCount > $maxConsumers) {
  exit;
}

Beenden Sie das Consumer-Skript einfach, wenn die maximale Anzahl an Consumern erreicht ist. Der große Vorteil von PHP ist, dass es bei Fehlern oder Exceptions kontrolliert abbricht. Das bedeutet, dass PHP-Consumer bei Problemen nicht in eine Sackgasse laufen. Tritt beim Konsumieren einer Nachricht ein Fehler auf (oder eine Exception, das liegt an Ihnen), stoppt der PHP-Prozess und damit auch das Konsumieren. Sie müssen weder in PHP noch in RabbitMQ zusätzlich aufräumen. Warten Sie einfach auf den nächsten Cron-Lauf bzw. die nächste geplante Aufgabe, und ein neuer, frischer Consumer wird gestartet.

Natürlich müssen Sie beim Exception-/Fehler-Handling noch etwas mehr tun, und all das sollte sauber in Klassen gekapselt werden und so weiter. Was auch immer zu Ihren Anforderungen passt.

Consumer automatisch erneuern

Um zu vermeiden, dass Consumer zu lange laufen, können Sie sie nach einer Weile herunterfahren. Innerhalb Ihres Consumer-Callbacks können Sie prüfen, wie lange der Consumer bereits läuft, und ihn nach einer definierten Zeit beenden:

// snippet of a possible consumer class

/**
 * Consume messages from queue
 *
 * @param PhpAmqpLibMessageAMQPMessage $message The message
 */
 public function consume(PhpAmqpLibMessageAMQPMessage $message) {
   // do whatever you have to do with your message
   $result = $this->executeWorker($message->body);

   if ($result === true) {
     // tell rabbitmq that message is completed
     $channel = $message->delivery_info['channel'];
     $channel->basic_ack($message->delivery_info['delivery_tag']);
   }

   // stop consuming when ttl is reached
   if ($this->start + $this->ttl) < time()) {
     $channel->basic_cancel($message->delivery_info['consumer_tag']);
   }
 }

Blockierte Ressourcen vermeiden

Um nicht in böse Blocking-Probleme zu geraten, sollten Sie stets vermeiden, Ressourcen an einen Consumer zu binden, die dort nicht benötigt werden. Beispielsweise brauchen Sie während des Konsumierens einer Nachricht vielleicht eine Datenbankverbindung. Wenn Sie sich mit einer Datenbank verbinden, bleibt die Verbindung für die gesamte Lebensdauer Ihres Consumers bestehen. Bei vielen Consumern öffnen Sie möglicherweise zu viele Verbindungen und blockieren andere bei der Nutzung der Datenbank.

Das ist ein Grund, warum wir den Consumer wirklich sehr, sehr leichtgewichtig halten. Um eine Nachricht zu verarbeiten, rufen wir mittels proc_open() ein weiteres PHP-Skript auf, um unnötige Bindungen innerhalb des Consumers selbst zu vermeiden. Die Bindungen bleiben beim Worker-Skript:

/**
 * Execute a worker in an external process
 *
 * @param string $body The job message body
 * @return boolean True on success
 */
protected function executeWorker($body) {
  // open a php process and call the worker.php script
  $pipes = array();
  $process = proc_open(
    '/usr/bin/php -d display_errors=stderr /path/to/worker.php',
    array(
      0 => array("pipe", "r"),
      1 => array("pipe", "w"),
      2 => array("pipe", "w")
    ),
    $pipes,
    sys_get_temp_dir(),
    null
  );

  if (is_resource($process)) {
    // write the message into worker.php stdin
    fwrite($pipes[0], $body);
    fclose($pipes[0]);

    // read errors from worker.php
    $stdErr = stream_get_contents($pipes[2]);
    fclose($pipes[2]);

    // if worker.php ends without errors, execution was successful
    if (proc_close($process) === 0 && empty($stdErr)) {
      return true;
    }
  }

  return false;
}

Letztlich ist es dank einiger PHP-Eigenheiten gar nicht so komplex, PHP-Consumer für die Verarbeitung von Nachrichten aus Queueing-Systemen zu betreiben. Das Gleiche funktioniert möglicherweise auch mit anderen Systemen wie zum Beispiel Gearman. Viel Spaß beim eigenen Ausprobieren.

An die Windows-Leute: Wie lösen Sie das? Lassen Sie es mich in den Kommentaren wissen.