当前位置:Gxlcms > PHP教程 > php发送数据到kafka实现代码

php发送数据到kafka实现代码

时间:2021-07-01 10:21:17 帮助过:31人阅读

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如何发送数据到kafka - kafka_version.png

php如何发送数据到kafka - receive.png

生产者代码(异步)

<?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\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);

生产者代码(同步)

<?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\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命令发送数据。

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 。

相关推荐:

linux下php扩展kafka的实例分享

使用 PHP 编写的 Kafka 客户端

kafka装配及Kafka-PHP扩展的使用

以上就是php发送数据到kafka实现代码的详细内容,更多请关注Gxl网其它相关文章!

人气教程排行