ホームページ >バックエンド開発 >PHPチュートリアル >PHP で書かれた Kafka クライアント

PHP で書かれた Kafka クライアント

小云云オリジナル: 2017-12-05 11:05:172501ブラウズ

Kafka-php は、純粋な PHP で書かれた Kafka クライアントを使用します。現在、プロジェクト v0.2.x は、オリジナルの v0.1.x と互換性がありません。 x Kafka PHP v0.1.x ドキュメントのドキュメントを参照できますが、v0.2.x に切り替えることをお勧めします。 v0.2.x は、kafka ブローカーと対話するために PHP 非同期実行を使用します。これは、v0.1.x よりも安定していて効率的です。PHP 言語で記述されているため、拡張機能をコンパイルせずに使用でき、アクセスとメンテナンスが軽減されます。費用がかかります。

インストール環境要件

PHPバージョンは5.5以上
Kafkaサーバーバージョンは0.8.0以上
消費モジュールKafkaサーバーバージョンは0.9.0以上である必要があります

インストール

Composer を使用してインストールします

次のように、composer の依存関係 nmred/kafka-php をプロジェクトのcomposer.json ファイルに追加するだけです:

{
	"require": {
		"nmred/kafka-php": "0.2.*"
	}
}

Produce

<?php
require &#39;../vendor/autoload.php&#39;;
date_default_timezone_set(&#39;PRC&#39;);
use Monolog\Logger;
use Monolog\Handler\StdoutHandler;
// Create the logger
$logger = new Logger(&#39;my_logger&#39;);
// Now add some handlers
$logger->pushHandler(new StdoutHandler());

// 设置生产相关配置，具体配置参数见 [Configuration](Configuration.md)
$config = \Kafka\ProducerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('10.13.4.159:9192');
$config->setBrokerVersion('0.9.0.1');
$config->setRequiredAck(1);
$config->setIsAsyn(false);
$config->setProduceInterval(500);
$producer = new \Kafka\Producer(function() {
	return array(
		array(
			'topic' => 'test',
			'value' => 'test....message.',
			'key' => 'testkey',
		),
	);
});
$producer->setLogger($logger);
$producer->success(function($result) {
	var_dump($result);
});
$producer->error(function($errorCode, $context) {
	var_dump($errorCode);
});
$producer->send();

Consumer

<?php
require &#39;../vendor/autoload.php&#39;;
date_default_timezone_set(&#39;PRC&#39;);
use Monolog\Logger;
use Monolog\Handler\StdoutHandler;
// Create the logger
$logger = new Logger(&#39;my_logger&#39;);
// Now add some handlers
$logger->pushHandler(new StdoutHandler());

$config = \Kafka\ConsumerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('10.13.4.159:9192');
$config->setGroupId('test');
$config->setBrokerVersion('0.9.0.1');
$config->setTopics(array('test'));
//$config->setOffsetReset('earliest');
$consumer = new \Kafka\Consumer();
$consumer->setLogger($logger);
$consumer->start(function($topic, $part, $message) {
	var_dump($message);
});

上記のコンテンツはPHP で書かれた Kafka クライアントチュートリアル。皆さんのお役に立てれば幸いです。