kafka安装和简单测试

安装kafka

下载

wget https://www-us.apache.org/dist/kafka/2.1.1/kafka_2.11-2.1.1.tgz

zookeeper.properties 是zookeeper的配置文件，默认端口号2181，可不做修改

server.properties 是kafka配置文件，将 zookeeper.connect 这行 改为自己的zookeeper地址和端口号

这样就可以哦

修改完成之后 返回kafka主目录

cd ..

运行zookeeper和kafka

bin/zookeeper-server-start.sh config/zookeeper.properties   运行zookeeper

不要关闭此窗口 再开一个新窗口 重新进入kafka目录

bin/kafka-server-start.sh config/server.properties     运行kafka

运行producer和consumer

跟上步操作一样 不要关闭窗口 重新开 重新进入kafka目录

创建一个topic为test

把ip和port改为自己zookeeper中的    咱们上边就是localhost:2181

bin/kafka-topics.sh --create --zookeeper ip:port --replication-factor 1 --partitions 1 --topic test

我们这么运行

(bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

)      

运行producer

<?php

$conf = new RdKafka\Conf();

$conf->setDrMsgCb(function ($kafka, $message) {

    file_put_contents("./c_dr_cb.log", var_export($message, true), FILE_APPEND);

});

$conf->setErrorCb(function ($kafka, $err, $reason) {

    file_put_contents("./err_cb.log", sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason).PHP_EOL, FILE_APPEND);

});

//设置消费组

$conf->set('group.id', 'myConsumerGroup');

$rk = new RdKafka\Consumer($conf);

$rk->addBrokers("127.0.0.1");

$topicConf = new RdKafka\TopicConf();

$topicConf->set('request.required.acks', 1);

//在interval.ms的时间内自动提交确认、建议不要启动

//$topicConf->set('auto.commit.enable', 1);

$topicConf->set('auto.commit.enable', 0);

$topicConf->set('auto.commit.interval.ms', 100);

// 设置offset的存储为file

//$topicConf->set('offset.store.method', 'file');

// 设置offset的存储为broker

 $topicConf->set('offset.store.method', 'broker');

//$topicConf->set('offset.store.path', __DIR__);

//smallest：简单理解为从头开始消费，其实等价于上面的 earliest

//largest：简单理解为从最新的开始消费，其实等价于上面的 latest

//$topicConf->set('auto.offset.reset', 'smallest');

$topic = $rk->newTopic("test", $topicConf);

// 参数1消费分区0

// RD_KAFKA_OFFSET_BEGINNING 重头开始消费

// RD_KAFKA_OFFSET_STORED 最后一条消费的offset记录开始消费

// RD_KAFKA_OFFSET_END 最后一条消费

$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);

//$topic->consumeStart(0, RD_KAFKA_OFFSET_END); //

//$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

while (true) {

    //参数1表示消费分区，这里是分区0

    //参数2表示同步阻塞多久

    $message = $topic->consume(0, 12 * 1000);

    if (is_null($message)) {

        sleep(1);

        echo "No more messages\n";

        continue;

    }

    switch ($message->err) {

        case RD_KAFKA_RESP_ERR_NO_ERROR:

            var_dump($message);

            break;

        case RD_KAFKA_RESP_ERR__PARTITION_EOF:

            echo "No more messages; will wait for more\n";

            break;

        case RD_KAFKA_RESP_ERR__TIMED_OUT:

            echo "Timed out\n";

            break;

        default:

            throw new \Exception($message->errstr(), $message->err);

            break;

    }

}

设置完启动，会接收到你刚刚输入的消息

成功了！