Kafka の概要と、PHP ベースの Kafka のインストールとテスト-PHPチュートリアル-php.cn

この記事の内容は、Kafka の導入と、PHP をベースにした Kafka のインストールとテストについてです。非常に詳細な内容です。困っている友人は参考にしてください。お役に立てれば幸いです。

はじめに

Kafka は、高スループットの分散パブリッシュ/サブスクライブメッセージングシステムです。

知っておくべき Kafka の役割

プロデューサー: プロデューサー。
消費者: 消費者。
topic: メッセージはトピックカテゴリに記録されます。Kafka ではメッセージシード (フィード) が分類されており、それぞれのメッセージの種類をトピックと呼びます。
broker: クラスター内で実行され、1 つまたは複数のサービスで構成されます。各サービスはブローカーと呼ばれます。コンシューマーは 1 つまたは複数のトピックをサブスクライブし、ブローカーからデータを取得してこれらを利用できます。リリースされたニュース。

クラシックモデル

1. トピックの下のパーティションはコンシューマの数より小さくすることはできません、つまり、トピックの下のコンシューマの数はパーティションより大きくすることはできません。これより大きくなると、自由時間が無駄になります
2 . トピックの下のパーティションは、異なるコンシューマグループのコンシューマによって同時に使用できます
3. トピックの下のパーティションは、そのトピックの 1 つのコンシューマによってのみ使用できます同じコンシューマグループ

Kafka の概要と、PHP ベースの Kafka のインストールとテスト

共通パラメータの説明

request.required.acks

Kafka プロデューサの ACK には 3 つのメカニズムがありますプロデューサーを初期化するときのプロデューサーコンフィギュレーションは、 request.required.acks のさまざまな値を介して設定できます。

0: これは、プロデューサーが、同期が完了したというブローカーからの確認を待たずに、次の (バッチ) メッセージの送信を続けることを意味します。このオプションでは、レイテンシーは最も低くなりますが、耐久性の保証は最も弱くなります (サーバーに障害が発生すると、リーダーが停止した場合に一部のデータが失われますが、プロデューサーはそれを認識せず、ブローカーは送信された情報を受信できません)。

1: これは、リーダーがデータを正常に受信して確認した後、プロデューサーが次のメッセージを送信することを意味します。このオプションを使用すると、クライアントがリクエストが成功したことをサーバーが確認するまで待機するため、耐久性が向上します (デッドリーダーに書き込まれたがまだレプリケートされていないメッセージのみが失われます)。

-1: これは、後続のコピーがデータの受信を確認するまで、プロデューサーが送信を完了しないことを意味します。
このオプションは最高の耐久性を提供します。少なくとも 1 つの同期されたレプリカが生きている限り、情報が失われないことが保証されます。

3 つのメカニズムでは、パフォーマンスが順番に低下し (プロデューサーのスループットが低下)、データの堅牢性が順番に向上します。

auto.offset.reset

1.early: オフセットを最も古いオフセットに自動的にリセットします
2.latest: オフセットを最新のオフセットに自動的にリセットします (デフォルト)
3. none: コンシューマグループが前のオフセットを見つけられない場合、コンシューマに例外がスローされます。
4. その他のパラメーター: コンシューマーに例外 (無効なパラメーター) をスローします

kafka のインストールと簡単なテスト

kafka をインストールします (インストールは必要ありません。解凍するだけです)

# 官方下载地址：http://kafka.apache.org/downloads
# wget https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.1/kafka_2.12-1.1.1.tgz
tar -xzf kafka_2.12-1.1.1.tgz
cd kafka_2.12-1.1.0

ログイン後にコピー

kafka サーバーの起動

# 需先启动zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

ログイン後にコピー

kafka クライアントテストの開始

# 创建一个话题，test话题2个分区
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic test
Created topic "test".

# 显示所有话题
bin/kafka-topics.sh --list --zookeeper localhost:2181
test

# 显示话题信息
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test    PartitionCount:2    ReplicationFactor:1    Configs:
    Topic: test    Partition: 0    Leader: 0    Replicas: 0    Isr: 0
    Topic: test    Partition: 1    Leader: 0    Replicas: 0    Isr: 0


# 启动一个生产者（输入消息）
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
[等待输入自己的内容 出现>输入即可]
>i am a new msg !
>i am a good msg ?

# 启动一个生产者（等待消息） 
# 注意这里的--from-beginning，每次都会从头开始读取，你可以尝试去掉和不去掉看下效果
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
[等待消息]
i am a new msg !
i am a good msg ?

ログイン後にコピー

kafka の php 拡張機能のインストール

git clone https://github.com/arnaud-lb/php-rdkafka.git
cd php-rdkafka
phpize
./configure
make all -j 5
sudo make install

vim [php]/php.ini
extension=rdkafka.so

ログイン後にコピー

php コードの練習

プロデューサー

<?php $conf = new RdKafka\Conf();
$conf->setDrMsgCb(function ($kafka, $message) {
    file_put_contents("./dr_cb.log", var_export($message, true).PHP_EOL, FILE_APPEND);
});
$conf->setErrorCb(function ($kafka, $err, $reason) {
    file_put_contents("./err_cb.log", sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason).PHP_EOL, FILE_APPEND);
});

$rk = new RdKafka\Producer($conf);
$rk->setLogLevel(LOG_DEBUG);
$rk->addBrokers("127.0.0.1");

$cf = new RdKafka\TopicConf();
$cf->set('request.required.acks', 0);
$topic = $rk->newTopic("test", $cf);

$option = 'qkl';
for ($i = 0; $i produce(RD_KAFKA_PARTITION_UA, 0, "qkl . $i", $option);
}


$len = $rk->getOutQLen();
while ($len > 0) {
    $len = $rk->getOutQLen();
    var_dump($len);
    $rk->poll(50);
}

ログイン後にコピー

プロデューサーの実行

php producer.php
# output

int(20)
int(20)
int(20)
int(20)
int(0)

# 你可以查看你刚才上面启动的消费者shell应该会输出消息
qkl . 0
qkl . 1
qkl . 2
qkl . 3
qkl . 4
qkl . 5
qkl . 6
qkl . 7
qkl . 8
qkl . 9
qkl . 10
qkl . 11
qkl . 12
qkl . 13
qkl . 14
qkl . 15
qkl . 16
qkl . 17
qkl . 18
qkl . 19

ログイン後にコピー

コンシューマー

<?php $conf = new RdKafka\Conf();
$conf->setDrMsgCb(function ($kafka, $message) {
    file_put_contents("./c_dr_cb.log", var_export($message, true), FILE_APPEND);
});
$conf->setErrorCb(function ($kafka, $err, $reason) {
    file_put_contents("./err_cb.log", sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason).PHP_EOL, FILE_APPEND);
});

//设置消费组
$conf->set('group.id', 'myConsumerGroup');

$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("127.0.0.1");

$topicConf = new RdKafka\TopicConf();
$topicConf->set('request.required.acks', 1);
//在interval.ms的时间内自动提交确认、建议不要启动
//$topicConf->set('auto.commit.enable', 1);
$topicConf->set('auto.commit.enable', 0);
$topicConf->set('auto.commit.interval.ms', 100);

// 设置offset的存储为file
//$topicConf->set('offset.store.method', 'file');
// 设置offset的存储为broker
 $topicConf->set('offset.store.method', 'broker');
//$topicConf->set('offset.store.path', __DIR__);

//smallest：简单理解为从头开始消费，其实等价于上面的 earliest
//largest：简单理解为从最新的开始消费，其实等价于上面的 latest
//$topicConf->set('auto.offset.reset', 'smallest');

$topic = $rk->newTopic("test", $topicConf);

// 参数1消费分区0
// RD_KAFKA_OFFSET_BEGINNING 重头开始消费
// RD_KAFKA_OFFSET_STORED 最后一条消费的offset记录开始消费
// RD_KAFKA_OFFSET_END 最后一条消费
$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
//$topic->consumeStart(0, RD_KAFKA_OFFSET_END); //
//$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

while (true) {
    //参数1表示消费分区，这里是分区0
    //参数2表示同步阻塞多久
    $message = $topic->consume(0, 12 * 1000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            var_dump($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "No more messages; will wait for more\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out\n";
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

ログイン後にコピー

サーバーのメタデータの表示 (トピック/パーティション/ブローカー)

<?php $conf = new RdKafka\Conf();
$conf->setDrMsgCb(function ($kafka, $message) {
    file_put_contents("./xx.log", var_export($message, true), FILE_APPEND);
});
$conf->setErrorCb(function ($kafka, $err, $reason) {
    printf("Kafka error: %s (reason: %s)\n", rd_kafka_err2str($err), $reason);
});

$conf->set('group.id', 'myConsumerGroup');

$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("127.0.0.1");

$allInfo = $rk->metadata(true, NULL, 60e3);

$topics = $allInfo->getTopics();

echo rd_kafka_offset_tail(100);
echo "--";

echo count($topics);
echo "--";


foreach ($topics as $topic) {

    $topicName = $topic->getTopic();
    if ($topicName == "__consumer_offsets") {
        continue ;
    }

    $partitions = $topic->getPartitions();
    foreach ($partitions as $partition) {
//        $rf = new ReflectionClass(get_class($partition));
//        foreach ($rf->getMethods() as $f) {
//            var_dump($f);
//        }
//        die();
        $topPartition = new RdKafka\TopicPartition($topicName, $partition->getId());
        echo  "当前的话题：" . ($topPartition->getTopic()) . " - " . $partition->getId() . " - ";
        echo  "offset：" . ($topPartition->getOffset()) . PHP_EOL;
    }
}

ログイン後にコピー