• 欢迎访问搞代码网站,推荐使用最新版火狐浏览器和Chrome浏览器访问本网站!
  • 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏搞代码吧

php kafka使用

php 搞代码 4年前 (2022-01-22) 44次浏览 已收录 0个评论

这篇文章主要介绍了php kafka使用 ,有着一定的参考价值,现在分享给大家,有需要的朋友可以参考一下

一、安装以及使用shell命令终端操作kafka
环境配置
1、下载最新版本的kafka: kafka_2.11-1.0.0.tgz
http://mirrors.shu.edu.cn/apache/kafka/1.0.0/kafka_2.11-1.0.0.tgz

2、配置,解压后进入config目录
2.1、配置zookeeper.properties
默认监听端口2181

2.2、配置server.properties
开启监听端口,比如监听本地
listeners=PLAINTEXT://10.168.1.99:9092

3、启动
3.1、启动zookeeper
./bin/zookeeper-server-start.sh config/zookeeper.properties

3.2、启动server
./bin/kafka-server-start.sh config/server.properties

此处启动server时报错内存不足,解决方案要么加大系统内存,要么调小kafka需求内存
我这是调小kafka启动所需内存大小
vim kafka-server-start.sh
把其中:export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
改为:export KAFKA_HEAP_OPTS="-Xmx512M -Xms512M"

4、启动测试的消费者,并监听topic的test
./bin/kafka-console-consumer.sh –zookeeper 10.168.1.99:2181 –topic test –from-beginning

5、启动测试的生产者,并监听topic的test
./bin/kafka-console-producer.sh –broker-list 10.168.1.99:9092 –topic test

此时,在生产者中输入数据后回车后,在消费者终端可以看到生产者产生的信息

二、使用PHP操作kafka

1、安装kafka的扩展php-rdkafka
1.1、在安装php-rdkafka之前,需要先安装librdkafka
git clone https://github.com/edenhill/librdkafka.git
./configure
make && make install

1.2、安装php-rdkafka
git clone https://github.com/arnaud-lb/php-rdkafka.git
cd php-rdkafka
phpize
./configure
make && make install

2、编写kafka生产者和消费者类

<?php/** * Kafka.php. * User: lvfk * Date: 2018/2/7 0007 * Time: 11:04 * Desc: Kafka服务 */namespace app\models;use yii\base\InvalidConfigException;class Kafka{    public $broker_list = '10.168.1.99:9092';//配置kafka,可以用逗号隔开多个kafka    public $topic = 'topic&#<strong>*本文来源gaodai#ma#com搞@代~码^网+</strong><strong>搞代gaodaima码</strong>39;;    public $partition = 0;    protected $producer = null;    protected $consumer = null;    public function __construct()    {        if (empty($this->broker_list)) {            throw new InvalidConfigException("broker not config");        }        $rk = new \RdKafka\Producer();        if (empty($rk)) {            throw new InvalidConfigException("producer error");        }        $rk->setLogLevel(LOG_DEBUG);        if (!$rk->addBrokers($this->broker_list)) {            throw new InvalidConfigException("producer error");        }        $this->producer = $rk;    }    /**     * 生产者     * @param array $messages     * @return mixed     */    public function send($messages = [])    {        $topic = $this->producer->newTopic($this->topic);        return $topic->produce(RD_KAFKA_PARTITION_UA, $this->partition, json_encode($messages));    }    /**     * 消费者     */    public function consumer($object, $callback){        $conf = new \RdKafka\Conf();        $conf->set('group.id', 0);        $conf->set('metadata.broker.list', $this->broker_list);        $topicConf = new \RdKafka\TopicConf();        $topicConf->set('auto.offset.reset', 'smallest');        $conf->setDefaultTopicConf($topicConf);        $consumer = new \RdKafka\KafkaConsumer($conf);        $consumer->subscribe([$this->topic]);        echo "waiting for messages.....\n";        while(true) {            $message = $consumer->consume(120*1000);            switch ($message->err) {                case RD_KAFKA_RESP_ERR_NO_ERROR:                    echo "message payload....";                    $object->$callback($message->payload);                    break;            }            sleep(1);        }    }}

3、配置

//配置kafka生产者'asyncLog' => [    'class' => '\\app\\models\\Kafka',    'broker_list' => '10.168.1.99:9092',    'topic' => 'asynclog']

4、在业务代码中生产消息
\Yii::$app->asyncLog->send(['this is IndexController,'.date('y-md H:i:s',time())]);

5、在yii的command中消费

5.1、编写KafkaController.php

<?php/** * @link http://www.yiiframework.com/ * @copyright Copyright (c) 2008 Yii Software LLC * @license http://www.yiiframework.com/license/ */namespace app\commands;use yii\console\Controller;/** * This command echoes the first argument that you have entered. * * This command is provided as an example for you to learn how to create console commands. * * @author Qiang Xue <[email protected]> * @since 2.0 */class KafkaController extends Controller{    /**     * This command echoes what you have entered as the message.     * @param string $message the message to be echoed.     */    public function actionConsume()    {        \Yii::$app->asyncLog->consumer($this, 'callback');    }    public function callback($message)    {        \Yii::info($message, 'testkafka');        \Yii::$app->log->setflushInterval(1);    }}

5.2、运行:
./yii kafka/consume

6、当第4步中生产数据后,在第5步的终端可以消费数据

相关推荐:

php发送数据到kafka实现代码

linux下php扩展kafka的实例分享

以上就是php kafka使用 的详细内容,更多请关注搞代码gaodaima其它相关文章!


搞代码网(gaodaima.com)提供的所有资源部分来自互联网,如果有侵犯您的版权或其他权益,请说明详细缘由并提供版权或权益证明然后发送到邮箱[email protected],我们会在看到邮件的第一时间内为您处理,或直接联系QQ:872152909。本网站采用BY-NC-SA协议进行授权
转载请注明原文链接:php kafka使用

喜欢 (0)
[搞代码]
分享 (0)
发表我的评论
取消评论

表情 贴图 加粗 删除线 居中 斜体 签到

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址