PHP と Apache Kafka を使用してリアルタイムストリーム処理を実装する方法-PHPチュートリアル-php.cn

PHP と Apache Kafka を使用してリアルタイムストリーム処理を実装する方法

WBOY

リリース： 2023-06-28 12:02:02

オリジナル

1354 人が閲覧しました

Apache Kafka は、高スループット、低遅延の分散パブリッシュ/サブスクライブメッセージングシステムです。高周波数、大容量のデータストリームを処理するリアルタイムストリーム処理システムのアーキテクチャで広く使用されています。この記事では、PHP と Apache Kafka を使用してリアルタイムストリーム処理を実装する方法を紹介します。

Apache Kafka のインストール

Apache Kafka の使用を開始する前に、まず Apache Kafka をインストールする必要があります。公式 Web サイトから Apache Kafka をダウンロードしてインストールすることも、オープンソースのインストールスクリプトを使用することもできます。ここでは、Apache Kafka が提供するバイナリバージョンを使用します。

Kafka プロデューサーの作成

次に、Kafka クラスターにデータをプッシュする Kafka プロデューサーを作成します。 PHP では、kafka-php 拡張機能を使用してこれを実現できます。

まず、kafka-php 拡張機能をダウンロードしてコンパイルする必要があります。詳しいインストール手順は、kafka-php の GitHub ページでご覧いただけます。インストールが完了したら、PHP コードで kafka-php 拡張機能を使用できるようになります。

次の例は、Kafka プロデューサを作成し、トピックにメッセージを送信する方法を示しています。

<?php
require_once('KafkaProducer.php');

$producer = new KafkaProducer('localhost:9092');
$producer->send([
    [
        'topic' => 'example-topic',
        'value' => 'Hello, Kafka!',
        'key' => 'key1'
    ]
]);
?>

ログイン後にコピー

上記のコードでは、最初に KafkaProducer オブジェクトを作成し、そのアドレスを指定します。カフカクラスタ。次に、send メソッドを通じてトピック (example-topic) にメッセージを送信しました。

送信されるメッセージは、メッセージの件名、内容、キーを含む配列です。キーを使用してメッセージをグループ化すると、Kafka クラスターが同じキーを持つメッセージを同じパーティションに分散できるようになります。

Kafka コンシューマーの作成

次に、Kafka クラスターからのデータを消費する Kafka コンシューマーを作成します。同様に、PHP では、kafka-php 拡張機能を使用してこれを実現できます。

<?php
require_once('KafkaConsumer.php');

$consumer = new KafkaConsumer('localhost:9092', 'example-group', ['example-topic']);
$consumer->consume(function($message) {
    echo $message->payload . "
";
});
?>

ログイン後にコピー

上記のコードでは、まず KafkaConsumer オブジェクトを作成し、Kafka クラスターのアドレス、コンシューマーグループ (グループ) の名前、および消費されるトピックを指定します。次に、consumption メソッドを通じてデータの消費を開始します。

consume メソッドは、Kafka クラスターから受信したメッセージを処理するためのパラメーターとしてコールバック関数を受け入れます。コールバック関数では、メッセージのコンテンツ (ペイロード) にアクセスできます。

Kafka コンシューマの作成時にコンシューマグループの名前を指定したことに注意してください。コンシューマグループは Kafka の重要な概念であり、メッセージをパーティションに分散するために使用されます。同じコンシューマグループ名を持つコンシューマは同じトピックを一緒に消費し、Kafka はそれらのコンシューマ間でメッセージを自動的に配布します。コンシューマグループの目的は、各メッセージが 1 回だけ消費されるようにすることです。

リアルタイムストリーム処理

ここで、上記の 2 つの例を組み合わせて、リアルタイムストリーム処理を実現します。 Kafka プロデューサーを作成し、トピックに定期的にメッセージを送信できます。次に、コールバック関数でトピックから受信したメッセージを処理する Kafka コンシューマーを作成できます。

次は、リアルタイムストリーム処理を示す例です:

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

ログイン後にコピー

上記のコードでは、最初に Kafka プロデューサと Kafka コンシューマを作成します。次に、トピックに定期的に乱数を送信し、トピックからのメッセージを消費するループに入ります。コンシューマコールバック関数では、受信した値をコンソールに出力します。

ここで説明するのは、単純なリアルタイムストリーム処理プロセスです。実際には、リアルタイムストリーム処理システムはより複雑で、複数のプロデューサーとコンシューマーが存在し、複数のトピックとパーティションが存在する場合があります。いずれにしても、PHP と Apache Kafka を使用すると、リアルタイムストリーム処理システムを簡単に構築し、高頻度で大容量のデータストリームを処理できます。

以上がPHP と Apache Kafka を使用してリアルタイムストリーム処理を実装する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。