Arbeiten mit Kafka in PHP: Bibliotheken, Verbindung und Codebeispiele

Arbeiten mit Kafka in PHP: Bibliotheken, Verbindung und Codebeispiele

\n\n

Apache Kafka ist eine verteilte Plattform für die Datenstromverarbeitung, die es ermöglicht, Nachrichtenströme in Echtzeit zu veröffentlichen, zu speichern und zu verarbeiten. Im PHP-Ökosystem wird für die Arbeit mit Kafka die Erweiterung php-rdkafka verwendet – ein Wrapper um die in C geschriebene Bibliothek librdkafka. In diesem Artikel werden wir uns ansehen, wie man das Tool installiert, einen Producer und einen Consumer schreibt und typische Anwendungsszenarien betrachtet.

\n\n

Was ist php-rdkafka?

\n\n

php-rdkafka ist die offizielle PHP-Erweiterung für die Arbeit mit Apache Kafka. Sie bietet eine leistungsstarke Schnittstelle zum Senden und Empfangen von Nachrichten. Die Erweiterung unterstützt alle wichtigen Kafka-Funktionen: Topics, Partitionen, Consumer-Gruppen, manuelle Commits, Komprimierung und vieles mehr.

\n\n

Hauptfunktionen:

\n

\n
• Erstellung von Produzenten (Producers) zum Senden von Nachrichten an Topics.
\n
• Erstellung von Verbrauchern (Consumers) zum Lesen von Nachrichten aus Topics.
\n
• Unterstützung von Consumer-Gruppen und automatischem Rebalancing.
\n
• Arbeit mit Partitionen und manuelle Verwaltung von Offsets.
\n
• Nachrichtenkomprimierung (gzip, snappy, lz4, zstd).
\n
• Asynchrones Senden und Batch-Verarbeitung.
\n

\n\n

Installation von php-rdkafka

\n\n

Die Installation der Erweiterung ist über PECL oder durch Kompilieren aus dem Quellcode möglich. Die empfohlene Methode ist über PECL:

\n\n

# Installation über PECL\npecl install rdkafka\n\n# Für Ubuntu/Debian installieren Sie zuerst librdkafka-dev\nsudo apt-get install librdkafka-dev\npecl install rdkafka\n\n# Fügen Sie die Erweiterung in php.ini hinzu\nextension=rdkafka.so

\n\n

Überprüfen Sie nach der Installation, ob die Erweiterung geladen ist:

\n\n

php -m | grep rdkafka

\n\n

Wenn Sie Composer verwenden, können Sie zusätzlich das Paket arnaud-lb/php-rdkafka für das automatische Laden installieren:

\n\n

composer require arnaud-lb/php-rdkafka

\n\n

Beispiel für einen Producer

\n\n

Der Producer sendet Nachrichten an ein bestimmtes Kafka-Topic. Betrachten wir ein einfaches Beispiel mit synchronem Senden:

\n\n

<?php\nrequire_once 'vendor/autoload.php';\n\n$conf = new RdKafka\\Conf();\n$conf->set('metadata.broker.list', 'localhost:9092');\n\n$producer = new RdKafka\\Producer($conf);\n$topic = $producer->newTopic('test_topic');\n\n// Nachricht senden\n$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Hallo, Kafka von PHP!');\n\n// Auf das Senden aller Nachrichten warten\n$producer->flush(1000);\necho "Nachricht gesendet\\n";

\n\n

Erläuterungen zum Code:

\n

\n
• metadata.broker.list – Adresse des Kafka-Brokers (standardmäßig localhost:9092).
\n
• RD_KAFKA_PARTITION_UA – automatische Partitionenauswahl.
\n
• flush(1000) – blockiert die Ausführung bis zum Senden aller Nachrichten (Timeout 1 Sekunde).
\n

\n\n

Beispiel für einen Consumer

\n\n

Der Consumer liest Nachrichten aus einem Topic. Wichtig ist die Angabe der Consumer-Gruppe und der Offset-Commit-Strategie:

\n\n

<?php\nrequire_once 'vendor/autoload.php';\n\n$conf = new RdKafka\\Conf();\n$conf->set('metadata.broker.list', 'localhost:9092');\n$conf->set('group.id', 'my_consumer_group');\n$conf->set('auto.offset.reset', 'earliest');\n\n$consumer = new RdKafka\\KafkaConsumer($conf);\n$consumer->subscribe(['test_topic']);\n\necho "Warte auf Nachrichten...\\n";\n\nwhile (true) {\n    $message = $consumer->consume(120 * 1000); // Timeout 2 Minuten\n    switch ($message->err) {\n        case RD_KAFKA_RESP_ERR_NO_ERROR:\n            echo "Nachricht empfangen: " . $message->payload . "\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__PARTITION_EOF:\n            echo "Ende der Partition erreicht\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__TIMED_OUT:\n            echo "Timeout\\n";\n            break;\n        defau