PHP扩展之kafka安装应用案例详解

[root@localhost ~]# java -version
java version "1.8.0_73"
Java(TM) SE Runtime Environment (build 1.8.0_73-b02)
Java HotSpot(TM) 64-Bit Server VM (build 25.73-b02, mixed mode)

[root@localhost ~]# vim /etc/profile

export JAVA_HOME=/usr/local/jdk/jdk1.8.0_73
export CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar
export PATH=$JAVA_HOME/bin:$PATH

[root@localhost ~]# source /etc/profile

[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/zookeeper-server-start.sh config/zookeeper.properties &

[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-server-start.sh config/server.properties &

[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

<?php
require 'vendor/autoload.php';

while (1) {
    $part = mt_rand(0, 1);
    $produce = \Kafka\Produce::getInstance('kafka0:2181', 3000);
    // get available partitions
    $partitions = $produce->getAvailablePartitions('topic_name');
    var_dump($partitions);
    // send message
    $produce->setRequireAck(-1);
    $produce->setMessages('topic_name', 0, array(date('Y-m-d H:i:s'));
   
    sleep(3);
}

require 'vendor/autoload.php';

$consumer = \Kafka\Consumer::getInstance('kafka0:2181');
$group = 'topic_name';
$consumer->setGroup($group);
$consumer->setFromOffset(true);
$consumer->setTopic('topic_name', 0);
$consumer->setMaxBytes(102400);
$result = $consumer->fetch();
print_r($result);
foreach ($result as $topicName => $partition) {
    foreach ($partition as $partId => $messageSet) {
    var_dump($partition->getHighOffset());
        foreach ($messageSet as $message) {
            var_dump((string)$message);
        }
    var_dump($partition->getMessageOffset());
    }
}