温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

PHP如何扩展kafka安装应用

发布时间:2021-09-07 13:16:14 来源:亿速云 阅读:185 作者:Yi 栏目:开发技术

这期内容当中小编将会给大家带来有关PHP如何扩展kafka安装应用,文章内容丰富且以专业的角度为大家分析和叙述,阅读完这篇文章希望大家可以有所收获。

一. 首先确认下jdk有没有安装

使用命令

[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)

如果有以上信息的话,就往下安装吧,有些可能是jdk对不上,那就装到对的上的。如果没有安装,就看一下下面的jdk安装方法:

http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

到这个地址下载jdk8版本,我下载的是jdk-8u73-linux-x64.tar.gz,然后解压到/usr/local/jdk/下。

然后打开/etc/profile文件

[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

这时jdk就生效了,可以使用 java -version验证下。

二. 接下来安装Kafka

1. 下载Kafka

到http://kafka.apache.org/downloads.html下载相应的版本,我使用的是kafka_2.9.1-0.8.2.2.tgz。

2. 下载完解压到你喜欢的目录

我是解压到 /usr/local/kafka/kafka_2.9.1-0.8.2.2

3. 运行默认的Kafka

启动Zookeeper server

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

启动Kafka server

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

运行生产者producer

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

运行消费者consumer

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

这样,在producer那边输入内容,consumer马上就能接收到。

4. 当有跨机的producer或consumer连接时

需要配置config/server.properties的host.name,要不然跨机的连不上。

三. Kafka-PHP扩展

使用了一圈,就https://github.com/nmred/kafka-php可以用。

我是使用composer安装的,以下是示例:

producer.php

<?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);
}

consumer.php

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());
    }
}

上述就是小编为大家分享的PHP如何扩展kafka安装应用了,如果刚好有类似的疑惑,不妨参照上述分析进行理解。如果想知道更多相关知识,欢迎关注亿速云行业资讯频道。

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI