上传安装包
解压安装包到指定目录
代码语言：javascript复制[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
代码语言：javascript复制[root@localhost kafka]# mv kafka_2.12-3.6.0 kafka修改config目录下的配置文件server.propertie
代码语言：javascript复制[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连接会出问题
配置环境变量
代码语言：javascript复制[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不得重复
代码语言：javascript复制broker.id=103
.......
.......
.......broker.id=104
.......
.......
.......启动集群
先启动zk
代码语言：javascript复制[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
代码语言：javascript复制[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进程了。
