PHP Academy · Lektion

Mit RabbitMQ in PHP arbeiten

Veröffentlichen und konsumieren Sie Nachrichten mit RabbitMQ.

Mit RabbitMQ in PHP arbeiten ist eine kostenlose PHP Academy-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des PHP Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der PHP Academy-Kurs umfasst insgesamt 4 Lektionen.

RabbitMQ in PHP

RabbitMQ ist ein Broker, der AMQP 0-9-1 spricht. Der etablierte Client für PHP ist php-amqplib/php-amqplib (reines PHP) oder die C-basierte Erweiterung ext-amqp. Diese Lektion verwendet php-amqplib, weil es überall dort läuft, wo Composer läuft.

Das grundlegende AMQP-Modell umfasst drei Akteure: Exchanges empfangen Nachrichten, Bindings leiten sie anhand eines Schlüssels weiter und Queues halten sie für Konsumenten bereit. Wenn Sie dieses Modell beherrschen, ist der Rest nur noch Detailarbeit.

Den Client installieren

Fügen Sie die Bibliothek mit Composer hinzu. Sie benötigt die Erweiterungen sockets und bcmath, die in CLI-PHP-Builds beide üblich sind.

composer require php-amqplib/php-amqplib
# Connection target, e.g. amqp://guest:guest@localhost:5672/

Das Exchange-/Queue-/Binding-Modell

Produzenten veröffentlichen Nachrichten in eine Exchange, niemals direkt in eine Queue. Der Exchange-Typ bestimmt das Routing:

direct – exakte Übereinstimmung mit dem Routing-Key.
topic – Wildcard-Muster wie order.*.eu.
fanout – Broadcast an alle gebundenen Queues.
headers – Übereinstimmung anhand von Header-Attributen.

Ein Binding verbindet eine Queue anhand eines Routing-Musters mit einer Exchange. Die Entkopplung der Produzenten von der Queue-Topologie ist der eigentliche Zweck der Exchange-Schicht.

Verbindung herstellen und deklarieren

Öffnen Sie eine Verbindung, rufen Sie einen Channel ab und deklarieren Sie Ihre Topologie. durable: true sorgt dafür, dass Exchange und Queue einen Neustart des Brokers überstehen. Deklarationen sind idempotent – wird eine bereits vorhandene Entität mit übereinstimmenden Argumenten deklariert, hat das keine Auswirkung.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$ch->exchange_declare('orders', 'topic', false, true, false);
$ch->queue_declare('orders.email', false, true, false, false);
$ch->queue_bind('orders.email', 'orders', 'order.created');

echo "Topology ready\n";
$ch->close();
$conn->close();

Eine Nachricht veröffentlichen

Verpacken Sie den Inhalt in eine AMQPMessage. Setzen Sie delivery_mode = 2, damit die Nachricht persistent ist – nur die Kombination aus dauerhafter Queue und persistenter Nachricht übersteht einen Neustart (eine der beiden Eigenschaften allein reicht nicht aus).

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$payload = json_encode(['orderId' => 42, 'total' => 19.90]);
$msg = new AMQPMessage($payload, [
    'content_type'  => 'application/json',
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
    'message_id'    => bin2hex(random_bytes(8)),
]);

$ch->basic_publish($msg, 'orders', 'order.created');
echo "Published\n";
$ch->close();
$conn->close();

Konsumieren mit manuellem Ack

Registrieren Sie mit basic_consume einen Callback. Übergeben Sie no_ack = false, damit Sie die Bestätigung selbst steuern. Bestätigen Sie erst, nachdem die Verarbeitung erfolgreich war; bei einem Fehler ermöglicht basic_nack mit requeue, dass RabbitMQ die Nachricht erneut zustellt.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->basic_qos(null, 10, null); // prefetch 10

$callback = function ($msg) {
    $data = json_decode($msg->getBody(), true);
    try {
        // ... do work ...
        $msg->ack();
    } catch (\Throwable $e) {
        $msg->nack(true); // requeue
    }
};

$ch->basic_consume('orders.email', '', false, false, false, false, $callback);
while ($ch->is_consuming()) {
    $ch->wait();
}

Prefetch und faire Zustellung

Standardmäßig verteilt RabbitMQ Nachrichten im Round-Robin-Verfahren an Consumer, ohne deren Auslastung zu berücksichtigen – bei einem langsamen Consumer stauen sich dadurch Aufgaben. basic_qos(null, prefetch, null) begrenzt die Anzahl der unbestätigten Nachrichten, die ein Consumer halten darf.

Setzen Sie prefetch bei rechenintensiven, ungleichmäßigen Aufgaben auf eine kleine Zahl (z. B. 1–10), damit der Broker neue Arbeit nur an Consumer mit freien Kapazitäten sendet. Das ist eine faire Zustellung.

Publisher Confirms

basic_publish kehrt sofort zurück und teilt Ihnen nicht mit, ob der Broker die Nachricht angenommen hat. Aktivieren Sie für zuverlässiges Publizieren Publisher Confirms: Der Broker sendet einen Ack, sobald die Nachricht sicher gespeichert oder geroutet wurde.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->confirm_select(); // enable confirms on this channel

$ch->set_ack_handler(fn($m) => print("confirmed\n"));
$ch->set_nack_handler(fn($m) => print("REJECTED\n"));

$ch->basic_publish(new AMQPMessage('hi'), 'orders', 'order.created');
$ch->wait_for_pending_acks(5.0); // block until confirmed or timeout

Dead-Lettering

Konfigurieren Sie das Argument x-dead-letter-exchange einer Queue, damit abgelehnte Nachrichten (nack ohne requeue) oder abgelaufene Nachrichten an einen DLX weitergeleitet werden. Kombinieren Sie dies bei Quorum-Queues mit x-delivery-limit, um die Anzahl der Wiederholungsversuche automatisch zu begrenzen.

<?php
use PhpAmqpLib\Wire\AMQPTable;

$args = new AMQPTable([
    'x-dead-letter-exchange'    => 'orders.dlx',
    'x-dead-letter-routing-key' => 'order.failed',
    'x-message-ttl'             => 60000, // ms before expiry
]);
// false=passive, true=durable, false=exclusive, false=autodelete, args
$ch->queue_declare('orders.email', false, true, false, false, false, $args);
$ch->queue_declare('orders.dead', false, true, false, false);
$ch->queue_bind('orders.dead', 'orders.dlx', 'order.failed');

Worker im Produktionsbetrieb ausführen

Einige in der Praxis bewährte Betriebsregeln für PHP-RabbitMQ-Worker:

Führen Sie Consumer als langlebige CLI-Prozesse unter einem Supervisor (systemd / Supervisor) aus, der sie nach dem Beenden neu startet.
PHP weist im Laufe der Zeit Speicherlecks auf – starten Sie den Worker nach N Nachrichten oder beim Überschreiten eines Speichergrenzwerts neu.
Senden Sie AMQP-Heartbeats und behandeln Sie SIGTERM für ein kontrolliertes Herunterfahren (die laufende Nachricht abschließen und anschließend stoppen).
Verwenden Sie für HA Quorum-Queues statt veralteter gespiegelter Queues.

Kontrolliertes Herunterfahren

Deployments senden SIGTERM. Ein naiver Worker beendet sich mitten in der Nachrichtenverarbeitung, wodurch eine erneute Zustellung erzwungen wird. Installieren Sie einen Signal-Handler, der ein Flag setzt; schließen Sie die aktuelle Nachricht ab, bestätigen Sie sie und verlassen Sie anschließend die Consume-Schleife sauber. pcntl_async_signals(true) ermöglicht es PHP, das Signal zwischen AMQP-Wartevorgängen zuzustellen.

<?php
pcntl_async_signals(true);
$running = true;
pcntl_signal(SIGTERM, function () use (&$running) {
    $running = false; // stop after the current message
    echo "SIGTERM: draining...\n";
});

while ($running && $ch->is_consuming()) {
    try {
        $ch->wait(null, false, 5); // wakes for signals
    } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) {
        // idle tick - loop and re-check $running
    }
}
$ch->close();
echo "Stopped cleanly\n";

Kurzer Check

Das Überstehen eines Neustarts des Brokers.

Zusammenfassung

Sie können jetzt eine echte RabbitMQ-Pipeline in PHP aufbauen:

Veröffentlichen Sie in einem Exchange; leiten Sie Nachrichten über Bindings an Queues weiter.
Eine dauerhafte Queue + persistente Nachricht übersteht Neustarts.
Konsumieren Sie mit manuellem Ack und stimmen Sie basic_qos prefetch für eine faire Zustellung ab.
Verwenden Sie Publisher Confirms für zuverlässige Sendungen und einen DLX für unverarbeitbare Nachrichten.
Betreiben Sie Worker unter einem Supervisor mit Heartbeats und kontrolliertem Herunterfahren.

Als Nächstes: Kafka, wenn Sie statt Aufgabenwarteschlangen Streaming mit hohem Durchsatz benötigen.

Häufig gestellte Fragen

Ist die Lektion „Mit RabbitMQ in PHP arbeiten“ kostenlos?

Ja — der vollständige Text von „Mit RabbitMQ in PHP arbeiten“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des PHP Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der PHP Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Mit RabbitMQ in PHP arbeiten“?

Veröffentlichen und konsumieren Sie Nachrichten mit RabbitMQ. Du übst PHP Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um PHP Academy zu starten?

Keine Vorkenntnisse erforderlich. PHP Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „Mit RabbitMQ in PHP arbeiten“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser PHP Academy-Lektion Code schreiben und ausführen?

Ja. Jede PHP Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

Warum asynchrone Nachrichtenübermittlung
Mit RabbitMQ in PHP arbeiten
Apache Kafka mit PHP
Ereignisgesteuerte Workflows erstellen

← Zurück zu PHP Academy