一、安裝以及使用shell命令終端操作kafka
環(huán)境配置
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、配置,解壓后進(jìn)入config目錄
2.1、配置zookeeper.properties
默認(rèn)監(jiān)聽(tīng)端口2181

2.2、配置server.properties
開(kāi)啟監(jiān)聽(tīng)端口，比如監(jiān)聽(tīng)本地
listeners=PLAINTEXT://10.168.1.99:9092

3、啟動(dòng)
3.1、啟動(dòng)zookeeper
./bin/zookeeper-server-start.sh config/zookeeper.properties

3.2、啟動(dòng)server
./bin/kafka-server-start.sh config/server.properties

此處啟動(dòng)server時(shí)報(bào)錯(cuò)內(nèi)存不足，解決方案要么加大系統(tǒng)內(nèi)存，要么調(diào)小kafka需求內(nèi)存
我這是調(diào)小kafka啟動(dòng)所需內(nèi)存大小
vim kafka-server-start.sh
把其中:export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
改為：export KAFKA_HEAP_OPTS="-Xmx512M -Xms512M"

4、啟動(dòng)測(cè)試的消費(fèi)者，并監(jiān)聽(tīng)topic的test
./bin/kafka-console-consumer.sh --zookeeper 10.168.1.99:2181 --topic test --from-beginning

5、啟動(dòng)測(cè)試的生產(chǎn)者，并監(jiān)聽(tīng)topic的test
./bin/kafka-console-producer.sh --broker-list 10.168.1.99:9092 --topic test

此時(shí)，在生產(chǎn)者中輸入數(shù)據(jù)后回車后，在消費(fèi)者終端可以看到生產(chǎn)者產(chǎn)生的信息

二、使用PHP操作kafka

1、安裝kafka的擴(kuò)展php-rdkafka
1.1、在安裝php-rdkafka之前，需要先安裝librdkafka
git clone http://github.com/edenhill/librdkafka.git
./configure
make && make install

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

2、編寫(xiě)kafka生產(chǎn)者和消費(fèi)者類

<?php/** * Kafka.php. * User: lvfk * Date: 2018/2/7 0007 * Time: 11:04 * Desc: Kafka服務(wù) */namespace app/models;use yii/base/InvalidConfigException;html' target='_blank'>class Kafka{    public $broker_list = '10.168.1.99:9092';//配置kafka，可以用逗號(hào)隔開(kāi)多個(gè)kafka    public $topic = 'topic';    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;    }    /**     * 生產(chǎn)者     * @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));    }    /**     * 消費(fèi)者     */    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生產(chǎn)者'asyncLog' => [    'class' => '//app//models//Kafka',    'broker_list' => '10.168.1.99:9092',    'topic' => 'asynclog']

4、在業(yè)務(wù)代碼中生產(chǎn)消息
/Yii::$app->asyncLog->send(['this is IndexController,'.date('y-md H:i:s',time())]);

5、在yii的command中消費(fèi)

5.1、編寫(xiě)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 <qiang.xue@gmail.com> * @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、運(yùn)行:
./yii kafka/consume


6、當(dāng)?shù)?步中生產(chǎn)數(shù)據(jù)后，在第5步的終端可以消費(fèi)數(shù)據(jù)

相關(guān)推薦：

php發(fā)送數(shù)據(jù)到kafka實(shí)現(xiàn)代碼

linux下php擴(kuò)展kafka的實(shí)例分享

以上就是php kafka使用 的詳細(xì)內(nèi)容，更多請(qǐng)關(guān)注 其它相關(guān)文章！

鄭重聲明：本文版權(quán)歸原作者所有，轉(zhuǎn)載文章僅為傳播更多信息之目的，如作者信息標(biāo)記有誤，請(qǐng)第一時(shí)間聯(lián)系我們修改或刪除，多謝。