Windows环境安装和配置Kafka
1、下载地址:
http://kafka.apache.org/downloads.html
清华镜像:https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/
如:
https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.0/kafka_2.13-2.4.0.tgz
http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.0/kafka_2.13-2.4.0.tgz
2、安装
由于Kafka依赖于Java JDK和zookeeper环境,所以首先应先按照顺序安装二者。
2.1、安装Java JDK(略过)
注意:需要设置JAVA_HOME和PATH的系统环境变量
如系统变量中增加变量名JAVA_HOME的变量值为D:\Net_Program\Net_Java
PATH中增加%JAVA_HOME%\bin和%JAVA_HOME%\jre\bin
2.2、安装zookeeper
2.2.1、下载地址:
官网地址:https://www.apache.org/dyn/closer.cgi/zookeeper/
清华镜像:https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/
如:https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.6.0/apache-zookeeper-3.6.0-bin.tar.gz
2.2.2、安装
将下载下来的Zookeeper压缩包解压到某目录下,如:D:\Net_Program\Net_Zookeeper
目录说明:
bin:操作kafka的可执行脚本,还包含windows下脚本
config:配置文件所在目录
libs:依赖库目录
site-docs:
2.2.3、修改配置文件
将解压目录conf中的zoo_sample.cfg复制一份,并重命名为zoo.cfg(什么要复制一份重命名为zoo.cfg?因为项目启动时默认去加载conf路径下的zoo.cfg作为启动的配置文件)
修改dataDir配置和增加dataLogDir配置,dataDir用于存放数据,dataLogDir用于存放日志,如下配置所示:
# 存放数据
dataDir=D:/Net_Program/Net_Zookeeper/data
# 存放日志
dataLogDir=D:/Net_Program/Net_Zookeeper/log
2.2.4、启动zookeeper
以管理员身份运行CMD,定位到安装目录下的bin目录,执行命令zkServer即可
或者手动双击zkServer.cmd
注:
zookeeper3.5.4版本需要JDK1.8以上版本支持
虽然最新版的Kafka中已经内置了zookeeper,但是不建议使用,最好全新部署一套
Zookeeper安装配置参考:https://blog.csdn.net/skh2015java/article/details/85014859
2.3、安装Kafka
将下载下来的Kafka压缩包解压到某目录下,如:D:\Net_Program\Net_Kafka
3、配置Kafka
3.1、修改server.properties
3.1.1、找到安装目录config中的server.properties文件,用记事本等工具打开,根据实际情况配置相关信息,如下所示对常用的配置进行修改或说明
# broker的全局唯一编号,不能重复
broker.id=0
# 用来监听链接的端口,producer或consumer将在此端口建立连接
port=9092
# 处理网络请求的线程数量
num.network.threads=3
# 用来处理磁盘IO的线程数量
num.io.threads=8
# 发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
# 接受套接字的缓冲区大小
socket.receive.buffer.bytes=102400
# 请求套接字的缓冲区大小
socket.request.max.bytes=104857600
# kafka消息存放的路径
log.dirs=/tmp/kafka-logs
# topic在当前broker上的分片个数
num.partitions=1
# 用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
# 默认消息的最大持久化时间,168小时,7天
log.retention.hours=168
# 日志文件中每个segment的大小,默认为1G
log.segment.bytes=1073741824
# 周期性检查文件大小的时间
log.retention.check.interval.ms=300000
# 日志清理是否打开,一般不用启用,启用的话可以提高性能
log.cleaner.enable=false
# zookeeper集群的地址,可以是多个,多个之间用逗号分割hostname1:port1,hostname2:port2,hostname3:port3
zookeeper.connect=localhost:2181
# zookeeper链接超时时间
zookeeper.connection.timeout.ms=6000
# partion buffer中,消息的条数达到阈值,将触发flush到磁盘
#log.flush.interval.messages=10000
# 消息buffer的时间,达到阈值,将触发flush到磁盘
#log.flush.interval.ms=1000
############################# 其他 #############################
# 消息保存的最大值20M
message.max.byte=20971520
# kafka保存消息的副本数,如果一个副本失效了,另一个还可以继续提供服务
default.replication.factor=2
# 取消息的最大直接数
replica.fetch.max.bytes=5242880
3.1.2、详细配置说明
参考:
apache kafka系列之server.properties配置文件参数说明:https://blog.csdn.net/lizhitao/article/details/25667831
Kafka安装教程(详细过程):https://blog.csdn.net/Poppy_Evan/article/details/79415460
3.2、启动Kafka服务
以管理员身份运行CMD,定位到安装目录,执行如下命令
.\bin\windows\kafka-server-start.bat .\config\server.properties
3.3、Kafka命令
3.3.1、创建主题
以管理员身份运行CMD,定位到安装目录,执行如下命令来创建一个名称为TestTheme1的主题
.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic TestTheme1
3.3.2、查看主题
以管理员身份运行CMD,定位到安装目录,执行如下命来查看创建的主题
.\bin\windows\kafka-topics.bat --zookeeper localhost:2181 --list
3.3.3、启动生产者producer
以管理员身份运行CMD(新开一个CMD窗体,即模拟生产者窗体),定位到安装目录,执行如下命
.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic TestTheme1
3.3.4、启动消费者consumer
以管理员身份运行CMD(新开一个CMD窗体,即模拟消费者窗体),定位到安装目录,执行如下命
.\bin\windows\kafka-console-consumer.bat --zookeeper localhost:2181 --topic TestTheme1 → 该命令可能会报错:zookeeper is not a recognized option
改用如下命令即可:
.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic TestTheme1 --from-beginning
3.3.5、测试发送和接收
切换到刚才创建的生产者窗体,输入内容回车,然后即可在消费者窗体看到生产者发送过来的消息内容
4、Kafka可视化工具
下载地址:http://www.kafkatool.com/download.html
如:http://www.kafkatool.com/download2/kafkatool_64bit.exe
下载完成安装即可
设置:
安装完成后,打开软件,Tools→Settings→Topics,将右侧的Default Content Types中的Key和Message选择为String,点击OK按钮
也可以鼠标点击具体的主题,右侧的属性(Properties)面板中,Content Types中的Key和Message选择为String,然后点击Update和Refresh即可
参考文章:
kafka可视化客户端工具(Kafka Tool)的基本使用:https://www.cnblogs.com/frankdeng/p/9452982.html
Kafka 可视化工具(Kafka Tool):https://www.cnblogs.com/miracle-luna/p/11299345.html
10、相关文档
Kafka安装与配置:https://blog.csdn.net/twypx/article/details/81257715