PHP implementiert die Open-Source-Echtzeit-Datenverarbeitung Kafka Stream-PHP-Tutorial-php.cn

Kafka Stream kann als Stream-Computing-Engine Echtzeitdaten schnell verarbeiten und sofort einsatzbereite verteilte Verarbeitungsfunktionen bereitstellen. Als beliebte Entwicklungssprache kann PHP seine guten Sprachfunktionen und Erweiterungsbibliotheken auch zur Implementierung der Kafka Stream-Datenverarbeitung nutzen.

In diesem Artikel wird erläutert, wie Sie mit PHP die Echtzeit-Datenverarbeitung von Kafka Stream entwickeln und anhand eines Beispiels demonstrieren, wie Sie mit PHP die vom Beobachtermodus generierten Echtzeitdaten analysieren.

Einführung in Kafka Stream

Kafka Stream ist eine schnelle und stabile Stream-Computing-Engine, die Echtzeitdaten zuverlässig verarbeiten kann und sofort verteilte Verarbeitungsfunktionen bereitstellt. Kafka Stream ist eine effiziente und flexible Datenverarbeitungsmethode, bei der Nachrichten in Kafka-Themen konsumiert, zur Verarbeitung an die Anwendung gesendet und die verarbeiteten Ergebnisse dann zurück an das Kafka-Thema gesendet werden.

Integration von PHP und Kafka Stream

In PHP können wir über die offiziell von Kafka Stream bereitgestellte Kafka-PHP-Bibliothek problemlos PHP-Anwendungen in Kafka Stream integrieren. Die folgenden Kafka Stream-Versionen werden von der Kafka-PHP-Bibliothek unterstützt:

Kafka 0.10.x
Kafka 0.11.x
Kafka 1.0.x
Kafka 1.1.x
Kafka. 2.0.x
Kaf ka 2 . 1.x
Kafka 2.2.x

Kafka-PHP-Bibliothek bietet die folgende Kernfunktionalität:

Produzent: Bietet die Möglichkeit, Kafka-Nachrichten zu erstellen und an ein bestimmtes Thema zu senden.
Consumer: Bietet die Möglichkeit, Kafka-Nachrichten zu konsumieren und unterstützt die automatische und manuelle Übermittlung.
Manager: Bietet die Möglichkeit, Kafka-Themen und -Partitionen zu erstellen und zu löschen.

Darüber hinaus bietet die Kafka-PHP-Bibliothek auch Unterstützung für die Swoole-Erweiterung von PHP. Die Leistung von PHP-Anwendungen kann durch den Einsatz der Swoole-Erweiterung weiter verbessert werden.

Beobachtermuster

Das Beobachtermuster ist ein Verhaltensentwurfsmuster, das eine Eins-zu-viele-Abhängigkeitsbeziehung zwischen Objekten definiert. Wenn sich der Zustand eines Objekts ändert, werden alle davon abhängigen Objekte benachrichtigt und automatisch aktualisiert. Das Beobachtermuster wird häufig in der Ereignisüberwachung, der UI-Programmierung und anderen Bereichen verwendet und kann eine effiziente Nachrichtenzustellung und -verarbeitung erreichen.

Implementieren Sie die Datenverarbeitung im Beobachtermodus von Kafka Stream.

Im Folgenden wird anhand eines Beispielcodes demonstriert, wie Sie mit PHP die Echtzeit-Datenverarbeitung von Kafka Stream entwickeln und den Beobachtermodus für die Datenanalyse anwenden.

4.1 Kafka-Produzenten implementieren

Zuerst müssen wir einen Produzenten erstellen, um Nachrichten zum Kafka-Thema zu senden. Das Folgende ist ein einfacher Beispielcode für einen Kafka-Produzenten:

<?php
require_once __DIR__ . '/vendor/autoload.php';

use RdKafkaConf;
use RdKafkaProducer;
use RdKafkaProducerTopic;

$conf = new Conf();
$conf->set('metadata.broker.list', 'kafka:9092');
$producer = new Producer($conf);
$topic = $producer->newTopic('topic1');
for ($i = 0; $i < 10; $i++) {
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message $i");
}
?>

Nach dem Login kopieren

Im obigen Code verwenden wir die von der RdKafka-Erweiterungsbibliothek bereitgestellte Producer-Klasse, um den Kafka-Produzenten zu implementieren und Nachrichten an das Kafka-Thema mit dem Namen „topic1“ zu senden. Bei der Implementierung des Kafka-Produzenten müssen wir darauf achten, die Verbindungskonfiguration des Kafka-Clusters einzurichten, um sicherzustellen, dass der Kafka-Cluster korrekt verbunden werden kann.

4.2 Kafka Consumer implementieren

Als nächstes müssen wir einen Kafka Consumer erstellen, um Daten aus dem Kafka-Thema zu konsumieren. Das Folgende ist ein einfacher Beispielcode für einen Kafka-Konsumenten:

<?php
require_once __DIR__ . '/vendor/autoload.php';

use RdKafkaConf;
use RdKafkaConsumer;
use RdKafkaTopicPartition;

$conf = new Conf();
$conf->set('metadata.broker.list', 'kafka:9092');
$consumer = new Consumer($conf);
$consumer->addBrokers('kafka:9092');
$topic = $consumer->newTopic('topic1');
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

while (true) {
    $message = $topic->consume(0, 1000);
    if ($message === null) {
        continue;
    }
    if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
        echo "Received message: {$message->payload}
";
    }
}

$consumer->close();
?>

Nach dem Login kopieren

Im obigen Code verwenden wir die von der RdKafka-Erweiterungsbibliothek bereitgestellte Consumer-Klasse, um den Kafka-Konsumenten zu implementieren, Daten aus dem Kafka-Thema namens „topic1“ zu konsumieren und Daten auszugeben zur Konsole. Beachten Sie, dass wir bei der Implementierung eines Kafka-Verbrauchers das Verbrauchsthema und den Offset festlegen müssen, um den Verbrauch zu starten.

4.3 Implementierung des Beobachtermusters

Wir können jetzt Daten aus dem Kafka-Thema nutzen, aber wie verwendet man das Beobachtermuster, um die Daten zu analysieren? Das Folgende ist ein einfacher Beispielcode für ein Beobachtermuster:

<?php
require_once __DIR__ . '/vendor/autoload.php';

use SplObserver;
use SplSubject;

class Producer implements SplSubject
{
    private array $observers = [];

    public function attach(SplObserver $observer):void
    {
        array_push($this->observers, $observer);
    }

    public function detach(SplObserver $observer):void
    {
        if (($key = array_search($observer, $this->observers, true)) !== false) {
            array_splice($this->observers, $key, 1);
        }
    }

    public function notify():void
    {
        foreach ($this->observers as $observer) {
            $observer->update($this);
        }
    }

    public function produce(string $message):void
    {
        echo "Producing message: {$message}
";
        $this->notify();
    }
}

class Consumer implements SplObserver
{
    public function update(SplSubject $subject):void
    {
        echo "Consuming message: {$subject}
";
    }
}

$producer = new Producer();
$producer->attach(new Consumer());
$producer->produce('Message 1');
?>

Nach dem Login kopieren

Im obigen Code definieren wir eine Hauptklasse namens Producer, implementieren die SplSubject-Schnittstelle und stellen die Beobachterverwaltungsmethoden zum Anhängen, Trennen, Benachrichtigen und Produzieren bereit. Wir haben außerdem eine Beobachterklasse namens Consumer definiert, die SplObserver-Schnittstelle implementiert und die Aktualisierungsmethode für die Verarbeitung von Nachrichten bereitgestellt. Schließlich haben wir eine Producer-Instanz erstellt und eine Consumer-Instanz als Beobachter angehängt, die Produce-Methode einmal ausgeführt und die Update-Methode des Consumers ausgelöst.

4.4 Implementieren Sie die Datenverarbeitung im Beobachtermodus von Kafka Stream.

Abschließend kombinieren wir die Codes in den vorherigen drei Schritten, um die Datenverarbeitung im Beobachtermodus von Kafka Stream zu implementieren. Das Folgende ist ein einfacher Beispielcode für die Kafka Stream-Datenverarbeitung:

<?php
require_once __DIR__ . '/vendor/autoload.php';

use RdKafkaConf;
use RdKafkaConsumer;
use RdKafkaProducer;
use RdKafkaTopicPartition;
use SplSubject;
use SplObserver;

class KafkaStream implements SplSubject
{
    private array $observers;
    private Conf $conf;
    private Producer $producer;
    private Consumer $consumer;

    public function __construct(string $bootstrap_servers)
    {
        $this->conf = new Conf();
        $this->conf->set('metadata.broker.list', $bootstrap_servers);
        $this->producer = new Producer($this->conf);
        $this->consumer = new Consumer($this->conf);
        $this->observers = [];
    }

    public function attach(SplObserver $observer):void
    {
        array_push($this->observers, $observer);
    }

    public function detach(SplObserver $observer):void
    {
        if (($key = array_search($observer, $this->observers, true)) !== false) {
            array_splice($this->observers, $key, 1);
        }
    }

    public function notify():void
    {
        foreach ($this->observers as $observer) {
            $observer->update($this);
        }
    }

    public function produce(string $message, string $topic):void
    {
        echo "Producing message: {$message}
";
        $this->producer->newTopic($topic)->produce(RD_KAFKA_PARTITION_UA, 0, $message);
        $this->notify();
    }

    public function consume(string $topic):void
    {
        $topic_partition = new TopicPartition($topic, 0);
        $this->consumer->assign([$topic_partition]);
        $this->consumer->seek($topic_partition, 0);

        while (true) {
            $message = $this->consumer->consume(0, 1000);
            if ($message === null) {
                continue;
            }
            if ($message->err !== RD_KAFKA_RESP_ERR_NO_ERROR) {
                echo "Error: {$message->errstr()}, exiting.
";
                break;
            }
            echo "Consuming message: {$message->payload}
";
        }

        $this->consumer->close();
    }
}

class Consumer implements SplObserver
{
    public function update(SplSubject $subject):void
    {
        echo "Processing message: {$subject}
";
    }
}

$bootstrap_servers = 'kafka:9092';
$kafka_stream = new KafkaStream($bootstrap_servers);
$kafka_stream->attach(new Consumer());
$kafka_stream->produce('Message 1', 'topic1');
$kafka_stream->consume('topic1');
?>

Nach dem Login kopieren

Im obigen Code definieren wir eine Klasse namens KafkaStream, implementieren die SplSubject-Schnittstelle und stellen die Kafka Stream-Verarbeitungskernmethoden „produzieren“ und „konsumieren“ sowie Beobachter-Verwaltungsmethoden bereit anhängen, abtrennen, benachrichtigen. Wir haben außerdem eine Beobachterklasse namens Consumer definiert, die SplObserver-Schnittstelle implementiert und die Aktualisierungsmethode für die Verarbeitung von Nachrichten bereitgestellt. Schließlich erstellen wir eine KafkaStream-Instanz und hängen eine Consumer-Instanz als Beobachter an, führen die Produce-Methode einmal aus, um eine Nachricht zu erzeugen, und konsumieren und verarbeiten die Nachricht in der Consumer-Methode.

Zusammenfassung

In diesem Artikel wird erläutert, wie Sie mit PHP die Echtzeit-Datenverarbeitung von Kafka Stream entwickeln und wie Sie das Beobachtermuster zur Analyse von Echtzeitdaten verwenden. Kafka Stream und das Observer-Muster sind eine leistungsstarke Kombination von Tools, die uns dabei helfen können, große Echtzeitdaten schnell zu verarbeiten und eine effiziente Nachrichtenzustellung und -verarbeitung zu erreichen.

Das obige ist der detaillierte Inhalt vonPHP implementiert die Open-Source-Echtzeit-Datenverarbeitung Kafka Stream. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!