Kafka-php使用纯粹的PHP编写的kafka客户端，目前支持0.8.x以上版本的Kafka，该项目v0.2.x和v0.1.x不兼容，如果使用原有的v0.1.x的可以参照文档 KafkaPHPv0.1.xDocument,不过建议切换到v0.2.x上。v0.2.x使用PHP异步执行的方式来和kafkabroker交互，较v0.1.x更加稳定高效,由于使用PHP语言编写所以不用编译任何的扩展就可以使用，降低了接入与维护成本。
安装环境要求PHP版本大于5.5
KafkaServer版本大于0.8.0
消费模块KafkaServer版本需要大于0.9.0
Installation使用Composer安装添加composer依赖 nmred/kafka-php 到项目的 composer.json 文件中即可，如：
{"require": {"nmred/kafka-php": "0.2.*"}}Produce<?phprequire '../vendor/autoload.php';date_default_timezone_set('PRC');use Monolog\Logger;use Monolog\Handler\StdoutHandler;// Create the logger$logger = new Logger('my_logger');// 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<?phprequire '../vendor/autoload.php';date_default_timezone_set('PRC');use Monolog\Logger;use Monolog\Handler\StdoutHandler;// Create the logger$logger = new Logger('my_logger');// 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);});BasicProtocol基础协议API调用方式见 Example
评论