kafka集群安装

---

前言

kafka版本：2.13-3.9.1
下载地址：https://kafka.apache.org/downloads
节点1: 192.168.0.114
节点2: 192.168.0.127
节点3: 192.168.0.128

---

一、下载

下载地址：https://kafka.apache.org/downloads

二、解压

在每台机器上解压到：/root/kafka_2.13-3.9.1

三、修改配置文件

每台机器都修改配置文件：

vim /root/kafka_2.13-3.9.1/config/kraft/server.properties

每个节点的node.id不一样，注意修改：controller.quorum.voters
节点1:

process.roles=broker,controller
node.id=1
controller.quorum.voters=1@192.168.0.114:9093,2@192.168.0.127:9093,3@192.168.0.128:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://192.168.0.114:9092,CONTROLLER://192.168.0.114:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka/data-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

节点2:

process.roles=broker,controller
node.id=2
controller.quorum.voters=1@192.168.0.114:9093,2@192.168.0.127:9093,3@192.168.0.128:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://192.168.0.127:9092,CONTROLLER://192.168.0.127:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka/data-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

节点3:

process.roles=broker,controller
node.id=3
controller.quorum.voters=1@192.168.0.114:9093,2@192.168.0.127:9093,3@192.168.0.128:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://192.168.0.128:9092,CONTROLLER://192.168.0.128:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka/data-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

四、创建数据目录

在每台机器上执行：

mkdir -p /opt/kafka/data-logs

五、初始化元数据（重要）

在节点1上执行：

CLUSTER_ID=$(/root/kafka_2.13-3.9.1/bin/kafka-storage.sh random-uuid)
echo $CLUSTER_ID

此时会输出一串id，类似：z5KQBvw2QV6UqMSaUDH5IQ
这个id是集群id，每台机器上都需要一样。

在每个节点上执行：

/root/kafka_2.13-3.9.1/bin/kafka-storage.sh format -t z5KQBvw2QV6UqMSaUDH5IQ -c /root/kafka_2.13-3.9.1/config/kraft/server.properties

此时会在/opt/kafka/data-logs中产生配置：meta.properties

六、启动

每个节点执行：

/root/kafka_2.13-3.9.1/bin/kafka-server-start.sh -daemon /root/kafka_2.13-3.9.1/config/kraft/server.properties

查看日志有没有报错。
有报错大概率是第五步有问题，如果解决不了，则在每台机器执行：rm -rf /opt/kafka/data-logs
再重新做第五步。

七、验证

在每个节点上执行：

# 查看当前 broker 状态
netstat -tnlp | grep 9092

在某个节点上创建topic：

# 创建 topic
/root/kafka_2.13-3.9.1/bin/kafka-topics.sh --create --topic test --bootstrap-server 192.168.0.114:9092 --partitions 3 --replication-factor 3

在某个节点上查看topic：

/root/kafka_2.13-3.9.1/bin/kafka-topics.sh --list --bootstrap-server 192.168.0.114:9092

八、生产消费消息

在某个节点上执行：

#生产消息
/root/kafka_2.13-3.9.1/bin/kafka-console-producer.sh --bootstrap-server 192.168.0.127:9092 --topic test

在某个节点上执行：

#消费消息
/root/kafka_2.13-3.9.1/bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.114:9092 --topic test --from-beginning

九、设置开机启动

在每个节点上执行：

vim /etc/systemd/system/kafka.service

将以下内容复制进去：

[Unit]
Description=Apache Kafka (KRaft mode)
After=network.target[Service]
Type=simple
User=root
Environment="JAVA_HOME=/root/jdk-17.0.11"
Environment="PATH=/root/jdk-17.0.11/bin:/usr/bin:/bin"
ExecStart=/root/kafka_2.13-3.9.1/bin/kafka-server-start.sh /root/kafka_2.13-3.9.1/config/kraft/server.properties
ExecStop=/root/kafka_2.13-3.9.1/bin/kafka-server-stop.sh
Restart=on-failure
LimitNOFILE=65536[Install]
WantedBy=multi-user.target

执行：

sudo systemctl daemon-reexec
sudo systemctl daemon-reload
sudo systemctl restart kafka
sudo systemctl status kafka -l

没问题后执行：

sudo systemctl enable kafka

---

总结

kafka集群安装