So implementieren Sie Echtzeit-Datenverarbeitung mit PHP und Kafka

In den letzten Jahren ist die Nachfrage nach Echtzeit-Datenverarbeitung immer weiter gestiegen. Kaltstart- und Batch-basierte Technologien können den Anforderungen der Echtzeit-Datenverarbeitung nicht mehr gerecht werden. Daher wenden sich immer mehr Unternehmen der Echtzeit-Datenverarbeitungstechnologie zu. In diesem Artikel wird erläutert, wie Sie mit PHP und Kafka eine Echtzeit-Datenverarbeitung implementieren.

Kafka ist eine verteilte Stream-Verarbeitungsplattform mit hohem Durchsatz, die ursprünglich von LinkedIn entwickelt wurde. Mit Kafka können neue Stream-Verarbeitungs-, Stapelverarbeitungs-, Nachrichtensysteme, Koordinationssysteme usw. erstellt werden.

PHP ist eine beliebte dynamische Programmiersprache, die häufig zum Erstellen von Internetanwendungen verwendet wird. Obwohl PHP nicht die erste Wahl für die Echtzeit-Datenverarbeitung ist, wird es häufig in der Webentwicklung und Datenverarbeitung eingesetzt.

Jetzt stellen wir die Schritte zur Verwendung von PHP und Kafka vor, um eine Echtzeit-Datenverarbeitung zu erreichen.

Schritt 1: PHP installieren und konfigurieren

Bevor wir mit der Echtzeit-Datenverarbeitung mit PHP beginnen, müssen wir die PHP-Umgebung installieren und notwendige PHP-Erweiterungen hinzufügen, wie z. B. Kafka-Erweiterungen und Redis-Erweiterungen.

Die Kafka-Erweiterung kann über diesen Link heruntergeladen und installiert werden. Kafka, pecl install kafka, um die Kafka-Erweiterung zu installieren.

Redis-Erweiterung Sie können die PHP-Redis-Erweiterung hier herunterladen und installieren, oder Sie können PECL verwenden, um sie zu installieren, Befehl: pecl install redis.

Nach der Installation und Konfiguration der PHP-Erweiterung können wir mit dem Schreiben von Echtzeit-Datenverarbeitungsprogrammen beginnen.

Schritt 2: Mit Kafka verbinden

In Kafka werden Kafka-Produzenten und Kafka-Konsumenten verwendet, um Datenströme zu verbinden, um Daten an die „Datenpipeline“ zu übertragen. In PHP können wir die von Kafka bereitgestellten Klassen KafkaProducer und KafkaConsumer verwenden und sie instanziieren, um eine Verbindung zu Kafka herzustellen.

Der Beispielcode lautet wie folgt:

<?php

$kafkaConf = new RdKafkaConf();
$kafkaConf->set('metadata.broker.list', 'localhost:9092');//设置kafka连接信息
$kafkaProducer = new RdKafkaProducer($kafkaConf);
$kafkaConsumer = new RdKafkaConsumer($kafkaConf);
$topic = $kafkaProducer->newTopic('sample');

?>

Schritt 3: Daten lesen

Wir können die KafkaConsumer-Klasse verwenden, um Echtzeit-Datenströme abzurufen. In Kafka gibt es das Konzept eines Streams, der den Datenfluss in eine oder mehrere Partitionen unterteilt. Jede Partition besteht aus einer Master-Partition und null oder mehreren Slave-Partitionen. In PHP können wir die Klasse KafkaConsumer verwenden, um ein Verbraucherobjekt zu instanziieren und eine oder mehrere Partitionen zu abonnieren, um Daten zu lesen.

Der Beispielcode lautet wie folgt:

<?php

$kafkaConf = new RdKafkaConf();
$kafkaConf->set('metadata.broker.list', 'localhost:9092');//设置kafka连接信息
$kafkaConsumer = new RdKafkaConsumer($kafkaConf);

$topicConf = new RdKafkaTopicConf();
$topicConf->set('auto.offset.reset', 'smallest');

$topic = $kafkaConsumer->newTopic('sample', $topicConf);

var_dump($topic->getMetadata(true, 10000));

$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
while (true) {
    $message = $topic->consume(0, 1000);
    if (null !== $message) {
        print_r($message->payload);
    }
}

?>

Schritt 4: Datenverarbeitung

Nach Erhalt der Daten können wir die Daten verarbeiten und im Speicher speichern. Wir können Redis verwenden, um Daten zu speichern und sicher aufzubewahren, indem wir die Daten regelmäßig und zum richtigen Zeitpunkt in der Datenbank aktualisieren.

Der Beispielcode lautet wie folgt:

<?php

$kafkaConf = new RdKafkaConf();
$kafkaConf->set('metadata.broker.list', 'localhost:9092');//设置kafka连接信息
$kafkaConsumer = new RdKafkaConsumer($kafkaConf);

$topicConf = new RdKafkaTopicConf();
$topicConf->set('auto.offset.reset', 'smallest');

$topic = $kafkaConsumer->newTopic('sample', $topicConf);

$redisClient = new Redis();
$redisClient->connect('127.0.0.1', 6379);

$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
while (true) {
    $message = $topic->consume(0, 1000);
    if (null !== $message) {
        $data = json_decode($message->payload);
        $redisClient->hMSet('my_data', [
            $data->key1 => $data->value1,
            $data->key2 => $data->value2,
        ]);
    }
}

?>

Schritt 5: Datensynchronisierung

Zuletzt müssen wir den Echtzeit-Datenstrom zurück in unsere Datenbank spülen. Wir können einen Timer und einen PHP-Prozess verwenden, um den Redis-Cache regelmäßig zurück in die Datenbank zu leeren.

Der Beispielcode lautet wie folgt:

<?php

$kafkaConf = new RdKafkaConf();
$kafkaConf->set('metadata.broker.list', 'localhost:9092');//设置kafka连接信息
$kafkaConsumer = new RdKafkaConsumer($kafkaConf);

$topicConf = new RdKafkaTopicConf();
$topicConf->set('auto.offset.reset', 'smallest');

$topic = $kafkaConsumer->newTopic('sample', $topicConf);

$redisClient = new Redis();
$redisClient->connect('127.0.0.1', 6379);

$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

$count = 0;
while (true) {
    $message = $topic->consume(0, 1000);
    if (null !== $message) {
        $data = json_decode($message->payload);
        $redisClient->hMSet('my_data', [
            $data->key1 => $data->value1,
            $data->key2 => $data->value2,
        ]);
        $count++;
        if ($count == 5) {
            $count = 0;
            $allData = $redisClient->hGetAll('my_data');
            //将数据更新到数据库中
            //...
        }
    }
}

?>

Fazit

In diesem Artikel haben wir vorgestellt, wie man Echtzeit-Datenverarbeitung mit PHP und Kafka implementiert. Verwenden Sie Kafka, um Echtzeitdaten einfach in Ihre Datenpipeline zu streamen, und verwenden Sie PHP, um die Daten zu verarbeiten und zu speichern. Wir verwenden Redis auch als Cache und In-Memory-Speicher für die Verarbeitung von Echtzeitdaten. Dieser Ansatz kann Caching- und Messaging-Lösungen problemlos ersetzen und bietet gleichzeitig eine höhere Leistung und Skalierbarkeit.

Das obige ist der detaillierte Inhalt vonSo implementieren Sie Echtzeit-Datenverarbeitung mit PHP und Kafka. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!