この記事の内容は、Kafka の導入と、PHP をベースにした Kafka のインストールとテストについてです。非常に詳細な内容です。困っている友人は参考にしてください。お役に立てれば幸いです。
はじめに
Kafka は、高スループットの分散パブリッシュ/サブスクライブ メッセージング システムです。
知っておくべき Kafka の役割
プロデューサー: プロデューサー 。
消費者: 消費者。
topic: メッセージはトピック カテゴリに記録されます。Kafka ではメッセージ シード (フィード) が分類されており、それぞれのメッセージの種類をトピックと呼びます。
broker: クラスター内で実行され、1 つまたは複数のサービスで構成されます。各サービスはブローカーと呼ばれます。コンシューマーは 1 つまたは複数のトピックをサブスクライブし、ブローカーからデータを取得してこれらを利用できます。リリースされたニュース。
クラシック モデル
1. トピックの下のパーティションはコンシューマの数より小さくすることはできません、つまり、トピックの下のコンシューマの数はパーティションより大きくすることはできません。これより大きくなると、自由時間が無駄になります
2 . トピックの下のパーティションは、異なるコンシューマ グループのコンシューマによって同時に使用できます
3. トピックの下のパーティションは、そのトピックの 1 つのコンシューマによってのみ使用できます同じコンシューマ グループ
共通パラメータの説明
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;
}
}ログイン後にコピー
関連する推奨事項:
kafka Kafka-PHP 拡張機能のインストールと使用、kafkakafka-php 拡張機能
Kafka アセンブリと Kafka-PHP 拡張機能の使用
以上がKafka の概要と、PHP ベースの Kafka のインストールとテストの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
