快捷導(dǎo)航

php測試kafka項目示例

更新時間：2020年02月06日 12:15:49 作者：李瓊濤

這篇文章主要介紹了php測試kafka項目,結(jié)合實例形式分析了kafka-php項目依賴的安裝、配置及測試使用技巧,需要的朋友可以參考下

本文實例講述了php測試kafka項目。分享給大家供大家參考，具體如下：

概述

Kafka是最初由Linkedin公司開發(fā)，是一個分布式、分區(qū)的、多副本的、多訂閱者，基于zookeeper協(xié)調(diào)的分布式日志系統(tǒng)（也可以當(dāng)做MQ系統(tǒng)），常見可以用于web/nginx日志、訪問日志，消息服務(wù)等等，Linkedin于2010年貢獻給了Apache基金會并成為頂級開源項目。

主要應(yīng)用場景是：日志收集系統(tǒng)和消息系統(tǒng)。

安裝kafka-php項目依賴

composer require nmred/kafka-php

produce.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(
  'topic' => 'test',
  'value' => $t,
  'key' => $t,
 ),
 );
});
$producer->success(function($result) {
 var_export($result);
});
$producer->error(function($errorCode) {
 var_dump('error', $errorCode);
});
$producer->send();

consumer.php

<?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($topic, $part, $message) {
 var_dump($message);
});

測試生產(chǎn)者