目录

1. 准备服务器

2. 选一台服务器配置kafka安装包

2.1 下载安装包

2.2 解压安装包

2.3 修改配置文件

3. 分发安装包到其他机器

4. 修改每台机器的broker.id

5. 配置环境变量

6. 启停kafka服务

6.1 启动kafak服务

6.2 停止kafka服务 

1. 准备服务器

                1.买几台云服务器 或者 自己搭建个虚拟机

                2.机器间已经做了免密配置

                         192.168.0.211 worker01

                         192.168.0.212 worker02

                         192.168.0.213 worker03

                3.安装了JDK1.8 且配置了环境变量  不明白的可以看这里哦💪🏻

                4.安装了zookeeper3.6.3 且配置了环境变量  不明白的可以看这里哦💪🏻

2. 选一台服务器配置kafka安装包

2.1 下载安装包

#下载kafka安装包
wget http://archive.apache.org/dist/kafka/3.3.2/kafka_2.12-3.3.2.tgz

2.2 解压安装包

#解压安装包
tar -zxvf kafka_2.12-3.3.2.tgz 

2.3 修改配置文件

配置文件位置:	kafka_2.12-3.3.2/config/server.properties

修改内容1:
#指定broker 在集群中的唯一编号，不能重复，只能是数字
broker.id=0	

修改内容2:
#指定 kafak数据存放的路径，路径不需要提前创建，kafka 自动帮你创建，可以 配置多个磁盘路径，路径与路径之间可以用"，"分隔
#建议在kafka安装包内部创建
log.dirs=/home/software/kafka_2.12-3.3.2/kafka-logs

修改内容3:
#配置连接 Zookeeper 集群地址(在 zk 根目录下创建/kafka，方便管理)
zookeeper.connect=worker01:2181,worker02:2181,worker03:2181/kafka

其他配置项说明，可根据实际情况按需配置

#broker 的全局唯一编号，不能重复，只能是数字
broker.id=0
#处理网络请求的线程数量 
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=worker01:2181,worker01:2181,worker01:2181/kafka

3. 分发安装包到其他机器

# 分发安装包到其他集群节点(worker02、worker03)
scp -r kafka_2.12-3.3.2 worker02:/home/software
scp -r kafka_2.12-3.3.2 worker03:/home/software

4. 修改每台机器的broker.id

tips : broker.id 不得重复，整个集群中唯一

1. 在 worker01 执行

#将 kafka_2.12-3.3.2/config/server.properties 中的 broker.id 设置为0
broker.id=0

2. 在 worker02 执行

2. 在 worker02 执行
#将 kafka_2.12-3.3.2/config/server.properties 中的 broker.id 设置为1
broker.id=1

3. 在 worker03 执行

3. 在 worker03 执行
#将 kafka_2.12-3.3.2/config/server.properties 中的 broker.id 设置为2
broker.id=2

5. 配置环境变量

有条件的可以 在/etc/profile.d目录下创建 worker_profile.sh 
或者直接修改 /etc/profile

#添加 kafka环境变量
export KAFKA_HOME=/home/software/kafka_2.12-3.3.2
export PATH=$KAFKA_HOME/bin:$PATH

tips: 记得执行下 source /etc/profile 让变量生效哦！

6. 启停kafka服务

6.1 启动kafak服务

tips:  先启动 Zookeeper 集群，然后再启动 Kafka服务

#1.启动kafak服务 (如果是集群模式，需要在每个节点上执行)
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

6.2 停止kafka服务 

tips：停止 Kafka 集群时，一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper集群

          因为 Zookeeper 集群当中记录着 Kafka 集群相关信息，Zookeeper 集群一旦先停止

          Kafka 集群就没有办法再获取停止进程的信息，只能手动杀死 Kafka 进程了

#停止kafak服务 (如果是集群模式，需要在每个节点上执行)
kafka-server-stop.sh