kafka仅仅是个小小的纽带。经常用于数据的发送及转移。在kafka官方的例子中，其实并没有php的相关实现版本。现在网上流传的kafka的相关php库，都是些编程爱好者们自己写的类库，所以就肯定不会有太统一的接口标准了。
下面以某个类库为例，展示相关的kafka的php扩展库使用。综合比较了几家kafka的php库，苏南大叔觉得下面的这个开源类库，nmred/kafka-php ，比较简洁方便一些。
先要安装composer的类库nmred/kafka-php 。
composer的基本使用问题，大家可以查看我的composer相关文章。 https://newsn.net/tag/composer/ 。
composer require "nmred/kafka-php" -vvv
当然，你也可以使用镜像加速下载。
{
"config": {
"secure-http": false,
"preferred-install": "dist",
"sort-packages": true
},
"repositories": {
"packagist": {
"type": "composer",
"url": "https://packagist.phpcomposer.com"
}
},
"require": {
"nmred/kafka-php": "v0.2.0.7"
}
}确定端口及topic，查看kafka版本号
我选择本地的端口是9092，topic是test1，同时查看我本地的kafka版本是0.11.0.0。这些在代码中都是要用到的。
立即学习“PHP免费学习笔记（深入）”；
生产者代码(异步）
pushHandler(new StdoutHandler());$config = \Kafka\ProducerConfig::getInstance();$config->setMetadataRefreshIntervalMs(10000);$config->setMetadataBrokerList('127.0.0.1:9092');$config->setBrokerVersion('0.11.0.0');$config->setRequiredAck(1);$config->setIsAsyn(false);$config->setProduceInterval(500);$producer = new \Kafka\Producer(function() {
return array(
array(
'topic' => 'test1', //注意对应topic
'key' => 'testkey',
'value' => 'test....message.',
),
);});// $producer->setLogger($logger);$producer->success(function($result) {
var_dump($result);});$producer->error(function($errorCode) {
var_dump($errorCode);});$producer->send(true);生产者代码(同步）
pushHandler(new StdoutHandler());$config = \Kafka\ProducerConfig::getInstance();$config->setMetadataRefreshIntervalMs(10000);$config->setMetadataBrokerList('127.0.0.1:9092');$config->setBrokerVersion('0.11.0.0');$config->setRequiredAck(1);$config->setIsAsyn(false);$config->setProduceInterval(500);$producer = new \Kafka\Producer();// $producer->setLogger($logger);for($i = 0; $i < 100; $i++) {
$result = $producer->send(array(
array(
'topic' => 'test1',
'value' => 'test1....message.',
'key' => 'key'.$i,
),
));
var_dump($result);}这两个生产者的代码，可以用下列shell命令接收。
kafka-console-consumer --bootstrap-server localhost:9092 --topic test1 --from-beginning
消费者代码
require '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('127.0.0.1:9092');$config->setGroupId('test');$config->setBrokerVersion('0.11.0.0');$config->setTopics(array('test1'));//$config->setOffsetReset('earliest');$consumer = new \Kafka\Consumer();//$consumer->setLogger($logger);$consumer->start(function($topic, $part, $message) {
var_dump($message);});这个消费者代码，可以通过下面的shell命令发送数据。
专为中小型企业定制的网络办公软件,富有竞争力的十大特性： 1、独创 web服务器、数据库和应用程序全部自动傻瓜安装，建立企业信息中枢 只需3分钟。 2、客户机无需安装专用软件，使用浏览器即可实现全球办公。 3、集成Internet邮件管理组件，提供web方式的远程邮件服务。 4、集成语音会议组件，节省长途话费开支。 5、集成手机短信组件，重要信息可直接发送到员工手机。 6、集成网络硬
kafka-console-producer --broker-list localhost:9092 --topic test1
值得特殊说明的是，这个消费者的代码，在网页里面执行也是可以的。页面会实时显示相关数据。估计php端是个无限长的endless状态。
当然，大家要记得开启zookeeper和kafka，才能做这些实验。相关kafka的安装问题，请点击这里查看。https://newsn.net/tag/kafka/ 。
本文中涉及的类库地址是：https://github.com/weiboad/kafka-php 。这个开源类库，似乎是国人作品，所以有个中文文档。见这里。https://github.com/weiboad/kafka-php/blob/master/README_CH.md 。
相关推荐：
