php如何发送数据到kafka?

如果本文对您有帮助，或者节约了您的时间，欢迎打赏瓶饮料，建立下友谊关系。
本博客不欢迎：各种镜像采集行为。请尊重原创文章内容，转载请保留链接信息。

PHP高手欢聚QQ群：218127186     逗逼虎开心聊QQ群：175454274

在苏南大叔给大家带来的大数据套装里面，kafka仅仅是个小小的纽带。经常用于数据的发送及转移。在kafka官方的例子中，其实并没有php的相关实现版本。现在网上流传的kafka的相关php库，都是些编程爱好者们自己写的类库，所以就肯定不会有太统一的接口标准了。

php如何发送数据到kafka? - kafka

PHP高手都在这里，就差你了

php如何发送数据到kafka?（图3-1）

下面以某个类库为例，展示相关的kafka的php扩展库使用。综合比较了几家kafka的php库，苏南大叔觉得下面的这个开源类库，nmred/kafka-php ，比较简洁方便一些。

composer安装nmred/kafka-php

先要安装composer的类库nmred/kafka-php。composer的基本使用问题，大家可以查看苏南大叔的composer相关文章。

Batch

composer require "nmred/kafka-php" -vvv

当然，你也可以使用镜像加速下载，加速下载版的composer.json如下图所示。

{
  "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

PHP高手都在这里，就差你了

php如何发送数据到kafka?（图3-2）

php如何发送数据到kafka? - receive

PHP高手都在这里，就差你了

php如何发送数据到kafka?（图3-3）

生产者代码(异步）

PHP

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

生产者代码(同步）

PHP

<?php
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\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命令接收。

Batch

kafka-console-consumer --bootstrap-server localhost:9092 --topic test1 --from-beginning

消费者代码

PHP

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

Batch

kafka-console-producer --broker-list localhost:9092 --topic test1

值得特殊说明的是，这个消费者的代码，在网页里面执行也是可以的。页面会实时显示相关数据。估计php端是个无限长的endless状态。

相关链接

这个开源类库在github上面叫做weiboad/kafka-php，似乎是国人作品，所以有个中文文档。

小结

当然，大家要记得开启zookeeper和kafka，才能做这些实验。相关kafka的安装问题，请点击这里查看。https://newsn.net/tag/kafka/ 。

  【php】秘籍文章入口，仅传授于有缘之人 php    kafka

欢迎转载传播本篇原创文章，转载请保留链接及作者信息。

本站采用创作共用版权协议, 要求署名、非商业用途和相同方式共享。
转载本站内容必须也遵循“署名-非商业用途-相同方式共享”的创作共用协议。
《程序如此灵动~》下所有原创文章，如被用于商业用途，请您按规定支付稿费。

Original url: Access
Created at: 2018-10-29 13:05:54
Category: default
Tags: none

未标明原创文章均为采集，版权归作者所有，转载无需和我联系，请注明原出处，南摩阿彌陀佛，知识，不只知道，要得到

请先后发表评论
  • 最新评论
  • 总共0条评论