kafka入门03——简单实战

目录

安装Java

安装Zookeeper

安装Kafka

生产与消费

---

主要是记录下Kafka的安装配置过程，前置条件需要安装jdk和zookeeper。

安装Java

1.Oracle官网下载对应jdk安装包

• 官网地址：Java Downloads | Oracle

• 好人分享了下载需要的oracle账号，传送门Oracle账号分享_oracle 账户分享-CSDN博客

2.将本地压缩包上传到虚拟机自定义路径，路径看诸君的习惯，敝人使的/usr/local/java

使用SSH远程连接工具FinalShell上传jdk压缩包（上传文件也看诸君喜好的SSH连接工具），FinalShell安装下载：【安装教程】SSH远程连接工具-FinalShell的安装_finalshell安装_Summer_may的博客-CSDN博客

3.解压缩，在压缩包路径下输入: tar -zxvf 上传的jdk压缩包名

tar -zvxf jdk-8u391-linux-x64.tar.gz

4.编辑环境变量，打开配置文件 vim /etc/profile 或者 vi /etc/profile。在文件最后添加：

export JAVA_HOME=/usr/local/java/jdk1.8.0_391export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jarexport PATH=$PATH:$JAVA_HOME/bin

• JAVA_HOME配置成自己的路径

• 按i进行编辑，完成后按esc退出编辑模式，“shift+：”输入“wq!”保存并退出

5.刷新全局配置使生效

cd / #退回到根目录. /etc/profile #环境变量配置刷新

最后检查java版本

java -version

安装Zookeeper

1.java确认安装过了，这里直接开始安装zookeeper。

多一句，自zk3.5.5版本以后，已编译的jar包，尾部有bin，应该使用的是apache-zookeeper-3.8.3-bin.tar.gz。避免报错：“找不到或无法加载主类 org.apache.zookeeper.server.quorum.QuorumPeerMain”。(图片来自网络引用)

官网地址：Apache Download Mirrors

2.上传压缩包并解压，参照上文中FinalShell的方式。

cd /usr/local/zookeeper/
tar -zvxf apache-zookeeper-3.8.3-bin.tar.gz 

3.重命名zoo_sample.cfg为zoo.cfg

# cp conf/zoo_sample.cfg conf/zoo.cfg

zookeeper的server启动脚本使用的配置文件名称是zoo.cfg，不改名会报错

4.进入bin目录，使用zkServer.sh启动zookeeper，如果报错“-bash: zkServer.sh: command not found”就使用“./zkServer.sh start”

# zkServer.sh start

5.查看启动状态

# zkServer.sh status

6.使用Cli验证

# zkCli.sh

简单操作：

1.创建节点 create /zkTest myData

2.查看节点 get /zkTest 或者 get -s /zkTest

3.删除节点 delete /zkTest

4.退出cli客户端 quit

安装Kafka

1.官网下载压缩包到本地。官网地址：https://kafka.apache.org/downloads

2.根据上文一样，使用远程连接工具FinalShell上传压缩包进行解压。

tar -zxvf  kafka_2.12-3.6.0.tgz

3.确认Kafka相关配置（kafka依赖Zookeeper先启动Zookeeper服务）

# cd /usr/local/kafka/kafka_2.12-3.5.0/config/
# vi server.properties 

4.进入bin目录使用kafka-server-start.sh通过config配置启动Zookeeper服务

[root@10 bin]# ./kafka-server-start.sh ../config/server.properties 

5.重新打开一个SSH连接，通过jps命令查看kafka启动状态

[root@10 kafka_2.12-3.5.0]# jps -l

生产与消费

1.进入kafka的bin目录，先来创建个topic验证下

[root@10 bin]# ./kafka-topics.sh --zookeeper localhost:2181/kafka --create --topic topic-test --replication-factor 3 --partitions 4

报错：Exception in thread "main" joptsimple.UnrecognizedOptionException: zookeeper is not a recognized option

原因是：Kafka 版本过高，命令不存在。

修改命令：

[root@10 bin]# ./kafka-topics.sh --bootstrap-server localhost:9092  --create --topic topic-test --replication-factor 3 --partitions 4

还是报错： ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 3 larger than available brokers: 1.

原因是：zookeeper使用的单机部署，只有一个broker。创建topic的命令中分区是4，副本是3，超出了数量限制。

修改命令：

[root@10 bin]# ./kafka-topics.sh --bootstrap-server localhost:9092  --create --topic topic-test --replication-factor 1 --partitions 4

创建topic成功

2.展示topic信息：

[root@10 bin]# ./kafka-topics.sh --bootstrap-server localhost:9092  --describe --topic topic-test

3.进入kafka的bin目录使用自带的kafka-console-consumer.sh脚本订阅主题topic-test。

[root@10 bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic-test

这个时候topic-test没有存入任何消息，所以脚本还不能消费任何消息。

4.打开一个新的SSH连接，进入kafka的bin目录使用自带的kafka-console-producer.sh脚本发送消息到主题topic-test。

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

发送“Hello kafka,”"Hello,test"

consumer连接窗口就能消费消息了。

到此，简单的kafka实战案例就结束了。