Kafka 安装部署与脚本

Kafka ZooKeeper 模式集群规划、安装部署、启停流程和常用运维脚本。

Kafka 安装部署与脚本

1、安装部署

1.1集群规划

kafka102	kafka103	kafka104
jdk	jdk	jdk
zookeeper	zookeeper	zookeeper
kafka	kafka	kafka

1.2 ZooKeeper 依赖说明

本文档使用 Kafka ZooKeeper 模式部署。该模式下，Kafka Broker 启动前必须先连接到可用的 ZooKeeper 集群。

ZooKeeper 在 Kafka 中负责保存和协调集群元数据，包括 Broker 注册信息、Controller 信息、Topic 元数据、分区副本状态和 ISR 变更通知。Kafka 的消息数据不写入 ZooKeeper，而是写入 Broker 本地的 log.dirs。

生产环境建议至少部署 3 个 ZooKeeper 节点，并保证半数以上节点存活。三节点 ZooKeeper 集群中，允许 1 个节点故障；如果只剩 1 个节点，ZooKeeper 集群无法形成多数派，Kafka 也会受到影响。

1.3 ZooKeeper 前置准备

以下示例与 Kafka 节点共用 kafka102、kafka103、kafka104 三台机器。实际生产环境也可以单独部署 ZooKeeper 节点，避免和 Broker 抢占磁盘 IO。

1）准备 ZooKeeper 配置

[root@kafka102 module]$ cd /opt/module/zookeeper-3.5.9/conf
[root@kafka102 conf]$ cp zoo_sample.cfg zoo.cfg
[root@kafka102 conf]$ vim zoo.cfg

zoo.cfg 示例：

# ZooKeeper 的基础时间单位，单位为毫秒。
tickTime=2000

# Follower 初次连接 Leader 时允许的最大 tick 数。
initLimit=10

# Leader 与 Follower 心跳、同步请求允许的最大 tick 数。
syncLimit=5

# 快照数据目录，目录下需要放置 myid 文件。
dataDir=/opt/module/zookeeper-3.5.9/zkdata

# 事务日志目录，生产环境建议放到独立磁盘。
dataLogDir=/opt/module/zookeeper-3.5.9/zkdata/logs

# 客户端连接端口，Kafka 的 zookeeper.connect 会连接该端口。
clientPort=2181

# 允许常用四字命令，便于安装验证和基础观测。
4lw.commands.whitelist=ruok,stat,mntr,conf

# 自动清理快照和事务日志，至少保留 10 个文件。
autopurge.snapRetainCount=10

# 每 24 小时执行一次自动清理，0 表示关闭。
autopurge.purgeInterval=24

# server.ID=主机名:集群通信端口:Leader 选举端口。
server.1=kafka102:2888:3888
server.2=kafka103:2888:3888
server.3=kafka104:2888:3888

关键参数说明：

参数	作用
`tickTime`	ZooKeeper 心跳和超时计算的基础单位
`initLimit`	Follower 启动后和 Leader 完成初始化同步的时间限制
`syncLimit`	Leader 与 Follower 正常同步请求的时间限制
`dataDir`	保存快照数据和 `myid` 文件
`dataLogDir`	保存事务日志，写入频繁，建议独立磁盘
`clientPort`	Kafka Broker 和客户端连接 ZooKeeper 的端口
`server.N`	ZooKeeper 集群节点列表，`N` 要和各节点 `myid` 一致

2）创建数据目录和 myid

每个 ZooKeeper 节点都要创建 dataDir，并写入唯一的 myid。myid 必须和 zoo.cfg 中的 server.N 对应。

# kafka102
[root@kafka102 zookeeper-3.5.9]$ mkdir -p zkdata/logs
[root@kafka102 zookeeper-3.5.9]$ echo 1 > zkdata/myid

# kafka103
[root@kafka103 zookeeper-3.5.9]$ mkdir -p zkdata/logs
[root@kafka103 zookeeper-3.5.9]$ echo 2 > zkdata/myid

# kafka104
[root@kafka104 zookeeper-3.5.9]$ mkdir -p zkdata/logs
[root@kafka104 zookeeper-3.5.9]$ echo 3 > zkdata/myid

3）启动并验证 ZooKeeper

[root@kafka102 zookeeper-3.5.9]$ bin/zkServer.sh start
[root@kafka103 zookeeper-3.5.9]$ bin/zkServer.sh start
[root@kafka104 zookeeper-3.5.9]$ bin/zkServer.sh start

查看状态：

[root@kafka102 zookeeper-3.5.9]$ bin/zkServer.sh status
[root@kafka103 zookeeper-3.5.9]$ bin/zkServer.sh status
[root@kafka104 zookeeper-3.5.9]$ bin/zkServer.sh status

正常情况下，三台节点中会有 1 个 leader，其余为 follower。

使用四字命令做基础检查：

[root@kafka102 zookeeper-3.5.9]$ echo ruok | nc kafka102 2181
imok

[root@kafka102 zookeeper-3.5.9]$ echo mntr | nc kafka102 2181

如果四字命令没有返回，先检查 zoo.cfg 中是否配置了 4lw.commands.whitelist。

4）创建 Kafka 根路径

Kafka 的 zookeeper.connect 示例使用了 /kafka 作为 chroot 路径：

zookeeper.connect=kafka102:2181,kafka103:2181,kafka104:2181/kafka

建议提前创建该路径，便于隔离 Kafka 元数据，也避免权限或版本差异导致自动创建失败。

[root@kafka102 zookeeper-3.5.9]$ bin/zkCli.sh -server kafka102:2181
[zk: kafka102:2181(CONNECTED) 0] create /kafka "kafka cluster metadata root"
[zk: kafka102:2181(CONNECTED) 1] ls /
[kafka, zookeeper]

1.4 官网

https://kafka.apache.org/

1.5集群部署

1）官方下载地址

http://kafka.apache.org/downloads.html

2）上传

[root@kafka102 software]$ rz
[root@kafka102 software]$ ls -l | grep -i kafka
-rw-r--r-- 1 kafka kafka 104001442 Aug 23 15:17 kafka_2.12-3.2.1.tgz

3）解压

[root@kafka102 software]$ tar -xzvf kafka_2.12-3.2.1.tgz -C /opt/module/

4）重命名

[root@kafka102 software]$ cd ../module/
[root@kafka102 module]$ mv kafka_2.12-3.2.1/ kafka-3.2.1

5）修改server.properties文件

[root@kafka102 module]$ cd kafka-3.2.1/
[root@kafka102 kafka-3.2.1]$ vim config/server.properties
#broker的全局唯一编号，不能重复，只能是数字。
broker.id=0
delete.topic.enable=true
auto.create.topics.enable=false
#kafka运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔
log.dirs=/opt/module/kafka-3.2.1/data
# 日志文件保留的最长时间，超时将被删除
log.retention.hours=168
# 或者保留时长（分钟）
log.retention.minutes=240
#配置连接Zookeeper集群地址（在zk根目录下创建/kafka，方便管理）
zookeeper.connect=kafka102:2181,kafka103:2181,kafka104:2181/kafka

zookeeper.connect 由两部分组成：

kafka102:2181,kafka103:2181,kafka104:2181：ZooKeeper 集群地址。
/kafka：Kafka 在 ZooKeeper 中使用的根路径，用于隔离 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-3.2.1/data
#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=kafka102:2181,kafka103:2181,kafka104:2181/kafka

6）分发

[root@kafka102 module]# rsync -r kafka-3.2.1 root@kafka104:/opt/module/
[root@kafka102 module]# rsync -r kafka-3.2.1 root@kafka103:/opt/module/

7）修改server.properties文件

修改kafka103/kafka104的server.properties文件

注：broker.id不得重复，整个集群中唯一

[root@kafka103 kafka-3.2.1]$ vim config/server.properties
broker.id=1

[root@kafka104 kafka-3.2.1]$ vim config/server.properties
broker.id=2

8）配置环境变量

[root@kafka102 kafka-3.2.1]$ vim /etc/profile.d/env.sh
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka-3.2.1
export PATH=$PATH:$KAFKA_HOME/bin

[root@kafka102 kafka-3.2.1]$ source /etc/profile

1.6 启动集群

1）启动zookeeper服务

[root@kafka102 zookeeper-3.5.9]$ zk.sh start

2）查看zookeeper状态

[root@kafka102 zookeeper-3.5.9]$ zk.sh status

确认 ZooKeeper 集群有 1 个 leader 和 2 个 follower 后，再启动 Kafka。

3）启动kafka服务

[root@kafka102 kafka-3.2.1]$ bin/kafka-server-start.sh -daemon config/server.properties
[root@kafka103 kafka-3.2.1]$ bin/kafka-server-start.sh -daemon config/server.properties
[root@kafka104 kafka-3.2.1]$ bin/kafka-server-start.sh -daemon config/server.properties

注意：配置文件的路径要能够到server.properties

4）查看kafka进程

[root@kafka102 kafka-3.2.1]$ xjps.sh
======root@kafka102======
16064 Kafka
15459 QuorumPeerMain
16155 Jps
======root@kafka103======
13962 QuorumPeerMain
14556 Kafka
14636 Jps
======root@kafka104======
13842 QuorumPeerMain
14498 Jps
14440 Kafka

5）查看自动生成的文件

[root@kafka102 kafka-3.2.1]$ ls -l logs/
total 92
-rw-rw-r-- 1 kafka kafka  7299 Aug 29 10:22 controller.log
-rw-rw-r-- 1 kafka kafka     0 Aug 29 10:22 kafka-authorizer.log
-rw-rw-r-- 1 kafka kafka     0 Aug 29 10:22 kafka-request.log
-rw-rw-r-- 1 kafka kafka  9939 Aug 29 10:22 kafkaServer-gc.log.0.current
-rw-rw-r-- 1 kafka kafka 29374 Aug 29 10:22 kafkaServer.out
-rw-rw-r-- 1 kafka kafka   172 Aug 29 10:22 log-cleaner.log
-rw-rw-r-- 1 kafka kafka 29374 Aug 29 10:22 server.log
-rw-rw-r-- 1 kafka kafka   728 Aug 29 10:22 state-change.log
[root@kafka102 kafka-3.2.1]$ ls -l datas/
total 4
-rw-rw-r-- 1 kafka kafka  0 Aug 29 10:22 cleaner-offset-checkpoint
-rw-rw-r-- 1 kafka kafka  0 Aug 29 10:22 log-start-offset-checkpoint
-rw-rw-r-- 1 kafka kafka 88 Aug 29 10:22 meta.properties
-rw-rw-r-- 1 kafka kafka  0 Aug 29 10:22 recovery-point-offset-checkpoint
-rw-rw-r-- 1 kafka kafka  0 Aug 29 10:22 replication-offset-checkpoint

6）验证 Kafka 写入 ZooKeeper 的元数据

Kafka Broker 启动后，可以通过 ZooKeeper 客户端观察 /kafka 路径下的元数据。

[root@kafka102 zookeeper-3.5.9]$ bin/zkCli.sh -server kafka102:2181
[zk: kafka102:2181(CONNECTED) 0] ls /kafka
[admin, brokers, cluster, config, controller, controller_epoch]

[zk: kafka102:2181(CONNECTED) 1] ls /kafka/brokers/ids
[0, 1, 2]

/kafka/brokers/ids 中能看到 0、1、2，说明三个 Broker 已经注册到 ZooKeeper。更多元数据说明可以参考 Kafka Broker 运维 文档。

7）关闭集群

[root@kafka102 kafka-3.2.1]$ bin/kafka-server-stop.sh
[root@kafka103 kafka-3.2.1]$ bin/kafka-server-stop.sh
[root@kafka104 kafka-3.2.1]$ bin/kafka-server-stop.sh

1.7 编写kafka脚本

在 /home/kafka/bin 目录下或者 /usr/local/bin 目录下创建常用运维脚本，并确保这些目录已经加入 PATH。

1）xjps.sh

xjps.sh 用于批量查看三台节点上的 Java 进程，确认 ZooKeeper 的 QuorumPeerMain 和 Kafka 的 Kafka 进程是否存在。

[root@kafka102 bin]$ sudo touch xjps.sh
[root@kafka102 bin]$ chmod 755 xjps.sh
[root@kafka102 bin]$ vim xjps.sh

#!/bin/bash
user=`whoami`

for((host=102; host<=104; host++)); do
    echo "======${user}@kafka${host}======"
    ssh ${user}@kafka${host} "source /etc/profile; jps"
done

执行示例：

[root@kafka102 bin]$ xjps.sh
======root@kafka102======
16064 Kafka
15459 QuorumPeerMain
16155 Jps
======root@kafka103======
13962 QuorumPeerMain
14556 Kafka
14636 Jps
======root@kafka104======
13842 QuorumPeerMain
14498 Jps
14440 Kafka

2）xkafka.sh

[root@kafka102 bin]$ sudo touch xkafka.sh
[root@kafka102 bin]$ chmod 755 xkafka.sh
[root@kafka102 bin]$ vim xkafka.sh

#!/bin/bash
user=`whoami`
kafka_home='/opt/module/kafka-3.2.1'

case $1 in
    "start")
        for((host=102; host<=104; host++)); do
            echo -e "---启动 ${user}@kafka${host} kafka---"
            ssh ${user}@kafka${host} "source /etc/profile; ${kafka_home}/bin/kafka-server-start.sh -daemon ${kafka_home}/config/server.properties"
        done
    ;;
    "stop")
        for((host=102; host<=104; host++)); do
            echo -e "---关闭 ${user}@kafka${host} kafka---"
            ssh ${user}@kafka${host} "source /etc/profile; nohup ${kafka_home}/bin/kafka-server-stop.sh ${kafka_home}/config/server.properties > /dev/null 2>&1 &"
            # 睡5秒钟
            sleep 5
        done
    ;;
    *)
        echo "你启动的姿势不对"
        echo "start 启动kafka集群"
        echo "stop  停止kafka集群"
    ;;
esac

注意：停止Kafka集群时，一定要等Kafka所有节点进程全部停止后再停止Zookeeper集群。因为Zookeeper集群当中记录着Kafka集群相关信息，Zookeeper集群一旦先停止，Kafka集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。

3）xkafka-topic-list.sh

[root@kafka102 bin]$ sudo touch xkafka-topic-list.sh
[root@kafka102 bin]$ sudo chown kafka:kafka xkafka-topic-list.sh
[root@kafka102 bin]$ chmod 755 xkafka-topic-list.sh
[root@kafka102 bin]$ vim xkafka-topic-list.sh

#!/bin/bash
kafka_home=/opt/module/kafka-3.2.1
# low version
#${kafka_home}/bin/kafka-topics.sh --zookeeper kafka102:2181,kafka103:2181,kafka104:2181 --list
# higit version
${kafka_home}/bin/kafka-topics.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --list

4）xkafka-topic-create.sh

[root@kafka102 bin]$ sudo touch xkafka-topic-create.sh
[root@kafka102 bin]$ chmod 755 xkafka-topic-create.sh
[root@kafka102 bin]$ vim xkafka-topic-create.sh

#!/bin/bash
kafka_home=/opt/module/kafka-3.2.1
#${kafka_home}/bin/kafka-topics.sh --zookeeper kafka102:2181,kafka103:2181,kafka104:2181 --create --replication-factor 2 --partitions 6 --topic $1
${kafka_home}/bin/kafka-topics.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --create --replication-factor 2 --partitions 6 --topic $1

5）xkafka-consume.sh

[root@kafka102 bin]$ sudo touch xkafka-consume.sh
[root@kafka102 bin]$ sudo chown kafka:kafka xkafka-consume.sh
[root@kafka102 bin]$ chmod 755 xkafka-consume.sh
[root@kafka102 bin]$ vim xkafka-consume.sh
#!/bin/bash
kafka_home=/opt/module/kafka-3.2.1
# 低版本
#${kafka_home}/bin/kafka-console-consumer.sh --zookeeper kafka102:2181,kafka103:2181,kafka104:2181 --topic $1
# 高版本
#${kafka_home}/bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --topic $1
# 从头开始消费
#${kafka_home}/bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --from-beginning --topic $1

case $1 in
    start)
        ${kafka_home}/bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --from-beginning --topic $2
    ;;
    stop)
        ${kafka_home}/bin/kafka-console-consumer.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --topic $2
    ;;
    *)
        echo "你启动的姿势不对"
        echo "  start  从头开始消费，包含历史数据"
        echo "  stop   从当前时刻开始消费"
    ;;
esac

6）xkafka-producer.sh

[root@kafka102 bin]$ sudo touch xkafka-producer.sh
[root@kafka102 bin]$ sudo chown kafka:kafka xkafka-producer.sh
[root@kafka102 bin]$ chmod 755 xkafka-producer.sh
[root@kafka102 bin]$ vim xkafka-producer.sh
#!/bin/bash
kafka_home=/opt/module/kafka-3.2.1
#${kafka_home}/bin/kafka-console-producer.sh --broker-list kafka102:9092,kafka103:9092,kafka104:9092 --topic $1
${kafka_home}/bin/kafka-console-producer.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --topic $1

7）xkafka-topic-describe.sh

[root@kafka102 bin]$ sudo touch xkafka-topic-describe.sh
[root@kafka102 bin]$ sudo chown kafka:kafka xkafka-topic-describe.sh
[root@kafka102 bin]$ chmod 755 xkafka-topic-describe.sh
[root@kafka102 bin]$ vim xkafka-topic-describe.sh
#!/bin/bash
kafka_home=/opt/module/kafka-3.2.1
#${kafka_home}/bin/kafka-topics.sh --zookeeper kafka102:2181,kafka103:2181,kafka104:2181 --describe --topic $1
${kafka_home}/bin/kafka-topics.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --describe --topic $1

8）xkafka-topic-delete.sh

[root@kafka102 bin]$ sudo touch xkafka-topic-delete.sh
[root@kafka102 bin]$ sudo chown kafka:kafka xkafka-topic-delete.sh
[root@kafka102 bin]$ chmod 755 xkafka-topic-delete.sh
[root@kafka102 bin]$ vim xkafka-topic-delete.sh
#!/bin/bash
kafka_home=/opt/module/kafka-3.2.1
#${kafka_home}/bin/kafka-topics.sh --zookeeper kafka102:2181,kafka103:2181,kafka104:2181 --delete --topic $1
${kafka_home}/bin/kafka-topics.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --delete --topic $1

9）xkafka-topic-update-partition.sh

[root@kafka102 bin]$ sudo touch xkafka-topic-update-partition.sh
[root@kafka102 bin]$ sudo chown kafka:kafka xkafka-topic-update-partition.sh
[root@kafka102 bin]$ chmod 755 xkafka-topic-update-partition.sh
[root@kafka102 bin]$ vim xkafka-topic-update-partition.sh
#!/bin/bash
kafka_home=/opt/module/kafka-3.2.1
${kafka_home}/bin/kafka-topics.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --alter --topic $1 --partitions $2