php测试kafka项目示例

poser require nmred/kafka-php

<?php
require './vendor/autoload.php';
date_default_timezone_set('PRC');
$config = \Kafka\ProducerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('127.0.0.1:9092');
$config->setBrokerVersion('0.10.2.1');
$config->setRequiredAck(1);
$config->setIsAsyn(false);
$config->setProduceInterval(500);
$producer = new \Kafka\Producer(function() {
 $t = time();
 return array(
 array(
  'ic' => 'test',
  'value' => $t,
  'key' => $t,
 ),
 );
});
$producer->suess(function($result) {
 var_export($result);
});
$producer->error(function($errorCode) {
 var_dump('error', $errorCode);
});
$producer->send();

<?php
require './vendor/autoload.php';
date_default_timezone_set('PRC');
$config = \Kafka\ConsumerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('127.0.0.1:9092');
$config->setGroupId('test');
$config->setBrokerVersion('0.10.2.1');
$config->setTopics(array('test'));
$consumer = new \Kafka\Consumer();
$consumer->start(function($ic, $part, $message) {
 var_dump($message);
});

php produce.php

php consumer.php