Apache Kafka ist ein verteiltes Publish/Subscribe-Messagingsystem mit hohem Durchsatz und geringer Latenz. Es wird häufig in der Architektur von Echtzeit-Stream-Verarbeitungssystemen zur Verarbeitung hochfrequenter Datenströme mit großer Kapazität verwendet. In diesem Artikel wird erläutert, wie Sie mit PHP und Apache Kafka die Echtzeit-Stream-Verarbeitung implementieren.
- Apache Kafka installieren
Bevor wir Apache Kafka verwenden, müssen wir es zuerst installieren. Sie können Apache Kafka von der offiziellen Website herunterladen und installieren oder einige Open-Source-Installationsskripte verwenden. Hier verwenden wir die von Apache Kafka bereitgestellte Binärversion.
- Erstellen Sie einen Kafka-Produzenten
Als Nächstes erstellen wir einen Kafka-Produzenten zum Übertragen von Daten an den Kafka-Cluster. In PHP können wir dazu die Erweiterung kafka-php verwenden.
Zuerst müssen wir die Kafka-PHP-Erweiterung herunterladen und kompilieren. Detaillierte Installationsanweisungen finden Sie auf der GitHub-Seite von kafka-php. Nachdem die Installation abgeschlossen ist, können wir die Erweiterung kafka-php in unserem PHP-Code verwenden.
Das Folgende ist ein Beispiel, das zeigt, wie man einen Kafka-Produzenten erstellt und Nachrichten an ein Thema sendet:
<?php
require_once('KafkaProducer.php');
$producer = new KafkaProducer('localhost:9092');
$producer->send([
[
'topic' => 'example-topic',
'value' => 'Hello, Kafka!',
'key' => 'key1'
]
]);
?>Nach dem Login kopieren
Im obigen Code erstellen wir zunächst ein KafkaProducer-Objekt und geben die Adresse des Kafka-Clusters an. Anschließend haben wir über die Sendemethode eine Nachricht an das Thema (Beispielthema) gesendet.
Die gesendete Nachricht ist ein Array, das den Betreff, den Inhalt und den Schlüssel der Nachricht enthält. Schlüssel können zum Gruppieren von Nachrichten verwendet werden, sodass der Kafka-Cluster Nachrichten mit demselben Schlüssel in derselben Partition verteilen kann.
- Erstellen Sie einen Kafka-Konsumenten
Als Nächstes erstellen wir einen Kafka-Konsumenten zum Konsumieren von Daten aus dem Kafka-Cluster. In ähnlicher Weise können wir in PHP die Erweiterung kafka-php verwenden, um dies zu erreichen.
<?php
require_once('KafkaConsumer.php');
$consumer = new KafkaConsumer('localhost:9092', 'example-group', ['example-topic']);
$consumer->consume(function($message) {
echo $message->payload . "
";
});
?>Nach dem Login kopieren
Im obigen Code erstellen wir zunächst ein KafkaConsumer-Objekt und geben die Adresse des Kafka-Clusters, den Namen der Verbrauchergruppe und das zu konsumierende Thema an. Dann beginnen wir mit der Datennutzung über die Methode „consume“.
Die Consumer-Methode akzeptiert eine Rückruffunktion als Parameter für die Verarbeitung von Nachrichten, die vom Kafka-Cluster empfangen werden. In der Callback-Funktion können wir auf den Inhalt der Nachricht (Payload) zugreifen.
Beachten Sie, dass wir beim Erstellen des Kafka-Verbrauchers den Namen der Verbrauchergruppe angegeben haben. Verbrauchergruppen sind ein Schlüsselkonzept in Kafka und werden zum Verteilen von Nachrichten in Partitionen verwendet. Verbraucher mit demselben Verbrauchergruppennamen konsumieren gemeinsam dasselbe Thema und Kafka verteilt automatisch Nachrichten unter ihnen. Der Zweck der Verbrauchergruppe besteht darin, sicherzustellen, dass jede Nachricht nur einmal konsumiert wird.
- Echtzeit-Stream-Verarbeitung
Jetzt können wir die beiden oben genannten Beispiele kombinieren, um eine Echtzeit-Stream-Verarbeitung zu erreichen. Wir können einen Kafka-Produzenten erstellen und regelmäßig Nachrichten an das Thema senden. Anschließend können wir einen Kafka-Konsumenten erstellen, der vom Thema empfangene Nachrichten in einer Rückruffunktion verarbeitet.
Hier ist ein Beispiel, das die Stream-Verarbeitung in Echtzeit demonstriert:
<?php
require_once('KafkaProducer.php');
require_once('KafkaConsumer.php');
$producer = new KafkaProducer('localhost:9092');
$consumer = new KafkaConsumer('localhost:9092', 'example-group', ['example-topic']);
while (true) {
$producer->send([
[
'topic' => 'example-topic',
'value' => rand(0, 10),
'key' => 'key1'
]
]);
$consumer->consume(function($message) {
$value = $message->payload;
echo "Received $value
";
});
sleep(1);
}
?>Nach dem Login kopieren
Im obigen Code erstellen wir zunächst einen Kafka-Produzenten und einen Kafka-Konsumenten. Anschließend treten wir in eine Schleife ein, die regelmäßig eine Zufallszahl an das Thema sendet und Nachrichten vom Thema verbraucht. In der Consumer-Callback-Funktion geben wir den empfangenen Wert auf der Konsole aus.
Was hier demonstriert wird, ist ein einfacher Echtzeit-Stream-Verarbeitungsprozess. In der Realität kann ein Echtzeit-Stream-Verarbeitungssystem komplexer sein, es kann mehrere Produzenten und Verbraucher sowie mehrere Themen und Partitionen geben. Aber in jedem Fall kann PHP und Apache Kafka problemlos ein Echtzeit-Stream-Verarbeitungssystem aufbauen und hochfrequente, großvolumige Datenströme verarbeiten.
Das obige ist der detaillierte Inhalt vonSo implementieren Sie Echtzeit-Stream-Verarbeitung mit PHP und Apache Kafka. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
