PHP はオープンソースの Kafka Stream リアルタイムデータ処理を実装します-PHPチュートリアル-php.cn

Kafka Stream は、ストリームコンピューティングエンジンとして、リアルタイムデータを迅速に処理し、すぐに使える分散処理機能を提供します。人気の開発言語である PHP は、その優れた言語機能と拡張ライブラリを使用して、Kafka Stream データ処理を実装することもできます。

この記事では、PHP を使用して Kafka ストリームのリアルタイムデータ処理を開発する方法を紹介し、例を使用してオブザーバーモードによって生成されたリアルタイムデータを PHP を使用して分析する方法を示します。

Kafka Stream の概要

Kafka Stream は、リアルタイムデータを確実に処理し、すぐに使える分散型データを提供できる、高速で安定したストリームコンピューティングエンジンです。処理能力。 Kafka Stream は、Kafka トピック内のメッセージを消費し、処理のためにアプリケーションに送信し、処理された結果を Kafka トピックに送り返すことにより、効率的かつ柔軟なデータ処理方法です。

PHP と Kafka Stream の統合

PHP では、Kafka Stream によって公式に提供される Kafka-PHP ライブラリを通じて、PHP アプリケーションを Kafka Stream と簡単に統合できます。。 Kafka-PHP ライブラリでサポートされている Kafka Stream のバージョンは次のとおりです:

Kafka 0.10.x
Kafka 0.11.x
Kafka 1.0.x
Kafka 1.1 .x
Kafka 2.0.x
Kafka 2.1.x
Kafka 2.2.x

Kafka-PHP ライブラリは、次のコア機能を提供します。 :

Producer: Kafka メッセージを生成し、指定されたトピックに送信する機能を提供します。
コンシューマ: Kafka メッセージを消費する機能を提供し、自動送信と手動送信をサポートします。
マネージャー: Kafka トピックとパーティションを作成および削除する機能を提供します。

さらに、Kafka-PHP ライブラリは PHP の Swoole 拡張機能のサポートも提供しており、Swoole 拡張機能を使用することで PHP アプリケーションのパフォーマンスをさらに向上させることができます。

Observer パターン

Observer パターンは、オブジェクト間の 1 対多の依存関係を定義する動作設計パターンです。それに依存するオブジェクトは通知され、自動的に更新されます。オブザーバーパターンは、イベント監視、UI プログラミング、その他の分野で広く使用されており、効率的なメッセージ配信と処理を実現できます。

Kafka Stream のオブザーバーモードデータ処理の実装

以下では、サンプルコードを使用して、PHP を使用して Kafka Stream のリアルタイムデータ処理を開発し、適用する方法を示します。データ分析用のオブザーバーモード。

4.1 Kafka プロデューサーの実装

まず、Kafka トピックにメッセージを送信するプロデューサーを作成する必要があります。以下は、簡単な Kafka プロデューサーのサンプルコードです。

<?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");
}
?>

ログイン後にコピー

上記のコードでは、RdKafka 拡張ライブラリによって提供されるプロデューサークラスを使用して、Kafka プロデューサを実装し、トピック内の 'topic1' という名前の Kafka にメッセージを送信します。 Kafka プロデューサーを実装するときは、Kafka クラスターが正しく接続できるように、Kafka クラスターの接続構成のセットアップに注意する必要があります。

4.2 Kafka コンシューマの実装

次に、Kafka トピックからのデータを消費する Kafka コンシューマを作成する必要があります。以下は、簡単な Kafka コンシューマサンプルコードです。

<?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();
?>

ログイン後にコピー

上記のコードでは、RdKafka 拡張ライブラリによって提供される Consumer クラスを使用して、Kafka コンシューマを実装し、「topic1」という名前の Kafka トピックからデータを消費します。データをコンソールに出力します。 Kafka コンシューマを実装するときは、消費トピックと消費を開始するオフセットを設定する必要があることに注意してください。

4.3 オブザーバーパターンの実装

これで、Kafka トピックからデータを利用できるようになりましたが、オブザーバーパターンを使用してデータを分析するにはどうすればよいでしょうか?以下は、単純なオブザーバーパターンのサンプルコードです。

<?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');
?>

ログイン後にコピー

上記のコードでは、Producer という名前のメインクラスを定義し、SplSubject インターフェイスを実装し、オブザーバー管理メソッドのアタッチ、デタッチ、通知、生成を提供します。また、Consumer という名前のオブザーバークラスを定義し、SplObserver インターフェイスを実装し、メッセージを処理するための update メソッドを提供しました。最後に、Producer インスタンスを作成し、Consumer インスタンスをオブザーバーとしてアタッチし、Produce メソッドを 1 回実行して、Consumer の Update メソッドをトリガーしました。

4.4 Kafka Stream のオブザーバーモードデータ処理を実装する

最後に、前の 3 つの手順のコードを組み合わせて、Kafka Stream のオブザーバーモードデータ処理を実装します。以下は、簡単な Kafka Stream データ処理のサンプルコードです。

<?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');
?>

ログイン後にコピー

上記のコードでは、KafkaStream という名前のクラスを定義し、SplSubject インターフェイスを実装し、Kafka Stream 処理のコアメソッドであるProduce および Consumer を提供します。オブザーバー管理メソッドとして、アタッチ、デタッチ、および通知が行われます。また、Consumer という名前のオブザーバークラスを定義し、SplObserver インターフェイスを実装し、メッセージを処理するための update メソッドを提供しました。最後に、KafkaStream インスタンスを作成し、Consumer インスタンスをオブザーバーとしてアタッチし、Produce メソッドを 1 回実行してメッセージを生成し、Consumer メソッドでメッセージを消費して処理しました。

概要

この記事では、PHP を使用して Kafka Stream のリアルタイムデータ処理を開発する方法を紹介し、オブザーバーパターンを使用してリアルタイムデータを分析する方法を示します。 Kafka Stream と Observer パターンは、大規模なリアルタイムデータを迅速に処理し、効率的なメッセージ配信と処理を実現するのに役立つ強力なツールの組み合わせです。

以上がPHP はオープンソースの Kafka Stream リアルタイムデータ処理を実装しますの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。