xxxxxxxxxx
log.dirs=/opt/kafka/logs
zookeeper.connect=node2:2181,node3:2181,node4:2181/kafka

属性了解：

xxxxxxxxxx
#broker的全局唯一编号，不能重复
broker.id=0
#删除topic功能,当前版本此配置默认为true，已从配置文件移除
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时7*24h将被删除，单位小时
log.retention.hours=168
#每个segment文件的大小，默认最大 1G
log.segment.bytes=1073741824
# 检查过期数据的时间，默认5分钟检查一次是否数据过期,单位ms
log.retention.check.interval.ms=300000
#配置连接zk集群地址(根目录下创建/kafka，否则在/下创建很多节点）
zookeeper.connect=node2:2181,node3:2181,node4:2181/kafka

6. 分发配置好的安装包到node3和node4上

   xxxxxxxxxx
   [root@node2 kafka]# cd /opt
   [root@node2 opt]# scp -r kafka/ node3:/opt
   [root@node2 opt]# scp -r kafka/ node4:/opt
   

7. 分别修改node3和node4上的配置文件server.properties,将broker.id分别改为1和2.

8. 配置node2,3,4环境变量,并生效

   xxxxxxxxxx
   [root@node2 opt]# vim /etc/profile
   # kafka的环境变量
   export KAFKA_HOME=/opt/kafka
   export PATH=$PATH:$KAFKA_HOME/bin
   [root@node2 opt]# source /etc/profile
   

9. 检查环境变量是否起作用：

   xxxxxxxxxx
   [root@node2 ~]# kafka-topics.sh --version
   3.0.1 (Commit:8e30984f43e64d8b)
   

10. 启动zk集群：zkServer.sh start

11. 启动kafka集群,三个节点都要执行：

    xxxxxxxxxx
    kafka-server-start.sh -daemon /opt/kafka/config/server.properties
    

    jps检查：

    xxxxxxxxxx
    [root@node2 opt]# jps
    3248 QuorumPeerMain
    3761 Jps
    3736 Kafka
    

12. 关闭kafka集群,三个节点都要执行：

    xxxxxxxxxx
    kafka-server-stop.sh