官方下载地址
http://kafka.apache.org/downloads.html
上传安装包
解压安装包到指定目录
[root@localhost kafka]# pwd /root/kafka [root@localhost kafka]# ll total 110744 -rw-r--r--. 1 root root 113400977 Oct 25 10:25 kafka_2.12-3.6.0.tgz [root@localhost kafka]# [root@localhost kafka]# [root@localhost kafka]# [root@localhost kafka]# [root@localhost kafka]# tar -xvzf kafka_2.12-3.6.0.tgz -C ./ ...... ...... ......
修改解压包名为kafka
[root@localhost kafka]# mv kafka_2.12-3.6.0 kafka
修改config目录下的配置文件server.propertie
[root@localhost kafka]# cd kafka [root@localhost kafka]# ls bin config libs LICENSE licenses NOTICE site-docs [root@localhost kafka]# cd config/ [root@localhost config]# ls connect-console-sink.properties connect-file-source.properties consumer.properties server.properties connect-console-source.properties connect-log4j.properties kraft tools-log4j.properties connect-distributed.properties connect-mirror-maker.properties log4j.properties trogdor.conf connect-file-sink.properties connect-standalone.properties producer.properties zookeeper.properties [root@localhost config]# ll server.properties -rw-r--r--. 1 root root 6896 Sep 29 12:56 server.properties [root@localhost config]# #broker的全局唯一编号，不能重复，只能是数字。 broker.id=102 #处理网络请求的线程数量 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运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔 log.dirs=/opt/module/kafka/datas #topic在当前broker上的分区个数 num.partitions=1 #用来恢复和清理data下数据的线程数量 num.recovery.threads.per.data.dir=1 # 每个topic创建时的副本数，默认时1个副本 offsets.topic.replication.factor=1 #segment文件保留的最长时间，超时将被删除 log.retention.hours=168 #每个segment文件的大小，默认最大1G log.segment.bytes=1073741824 # 检查过期数据的时间，默认5分钟检查一次是否数据过期 log.retention.check.interval.ms=300000 #配置连接Zookeeper集群地址（在zk根目录下创建/kafka，方便管理） zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka
建议也修改 listeners 和 advertised.listeners
不然非本机部署的外部应用通过物理IP连接会出问题
配置环境变量
[root@localhost profile.d]# cat /etc/profile.d/my_env.sh #KAFKA_HOME export KAFKA_HOME=/root/kafka/kafka export PATH=$PATH:$KAFKA_HOME/bin [root@localhost profile.d]# [root@localhost profile.d]# source /etc/profile [root@localhost profile.d]# [root@localhost profile.d]# [root@localhost profile.d]#
其他机器同上 - 修改配置文件中的brokerid
分别在hadoop103和hadoop104主机上修改配置文件server.properties中broker.id=103、broker.id=104
注：broker.id不得重复
broker.id=103 ....... ....... .......
broker.id=104 ....... ....... .......
启动集群
先启动zk
[root@localhost conf]# cd /root/kafka/ [root@localhost kafka]# cd kafka [root@localhost kafka]# ll total 68 drwxr-xr-x. 3 root root 4096 Sep 29 13:00 bin drwxr-xr-x. 3 root root 4096 Oct 25 11:27 config drwxr-xr-x. 2 root root 8192 Oct 25 10:29 libs -rw-r--r--. 1 root root 14973 Sep 29 12:56 LICENSE drwxr-xr-x. 2 root root 4096 Sep 29 13:00 licenses -rw-r--r--. 1 root root 28184 Sep 29 12:56 NOTICE drwxr-xr-x. 2 root root 44 Sep 29 13:00 site-docs [root@localhost kafka]# cd bin/ [root@localhost bin]# ./zookeeper-server-start.sh USAGE: ./zookeeper-server-start.sh [-daemon] zookeeper.properties [root@localhost bin]# ./zookeeper-server-start.sh -daemon ../config/zookeeper.properties [root@localhost bin]#
建议使用单独的zk
然后启动kafka
[root@localhost bin]# export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.382.b05-1.el7_9.x86_64/jre [root@localhost bin]# ./kafka-server-start.sh -daemon ../config/server.properties [root@localhost bin]#
停止
直接调用 kafka-server-stop.sh 即可
注意：停止Kafka集群时，一定要等Kafka所有节点进程全部停止后再停止Zookeeper集群。
因为Zookeeper集群当中记录着Kafka集群相关信息，Zookeeper集群一旦先停止，Kafka集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。
