部署kafka
集群规划
一般模式下，元数据在 zookeeper 中，运行时动态选举 controller，由controller 进行 Kafka 集群管理。kraft 模式架构下，不再依赖 zookeeper 集群，而是用三台 controller 节点代替 zookeeper，元数据保存在 controller 中，由 controller 直接进行 Kafka 集群管理。
| ip | 主机名 | 角色 | node id |
|---|---|---|---|
| 192.168.100.131 | kafka-1 | Broker，Controller | 1 |
| 192.168.100.132 | kafka-2 | Broker，Controller | 2 |
| 192.168.100.133 | kafka-3 | Broker，Controller | 3 |
1. 下载二进制包
# 3个节点上都要执行wget https://dlcdn.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgztar -zxf kafka_2.13-3.9.0.tgzmv kafka_2.13-3.9.0 /opt/kafkacd !$mkdir -p /data/kafka/apt-get updateapt-get install openjdk-21-jdk -yjava -version2. 修改配置文件
kafka 侦听器类型介绍：
PLAINTEXT：用于不加密的普通通信。 listeners=PLAINTEXT://:9092
SSL：用于加密通信，确保数据传输的安全性。 listeners=SSL://:9093
SASL_PLAINTEXT：在不加密的基础上，添加身份验证机制。listeners=SASL_PLAINTEXT://:9094
SASL_SSL：结合加密和身份验证，确保通信的机密性和完整性。listeners=SASL_SSL://:9095
CONTROLLER：用于 Kafka 集群控制器进行内部通信，管理 Broker 状态。listeners=CONTROLLER://:9096
EXTERNAL：专为外部客户端访问设计，通常用于跨网络的通信。listeners=EXTERNAL://:9097
# kafka-1vim /opt/kafka/config/kraft/server.properties# 节点ID，集群内唯一
node.id=1
# 集群地址信息
controller.quorum.voters=1@192.168.100.131:9093,2@192.168.100.132:9093,3@192.168.100.133:9093
# 侦听器名称、主机名和代理将向客户端公布的端口.(broker 对外暴露的地址)
advertised.listeners=PLAINTEXT://192.168.100.131:9092,CONTROLLER://192.168.100.131:9093
# kafka数据目录
log.dirs=/data/kafka# kafka-2vim config/kraft/server.properties# 节点ID，集群内唯一
node.id=2
# 集群地址信息
controller.quorum.voters=1@192.168.100.131:9093,2@192.168.100.132:9093,3@192.168.100.133:9093
# 侦听器名称、主机名和代理将向客户端公布的端口.(broker 对外暴露的地址)
advertised.listeners=PLAINTEXT://192.168.100.132:9092,CONTROLLER://192.168.100.132:9093
# kafka数据目录
log.dirs=/data/kafka# kafka-3vim config/kraft/server.properties
# 节点ID，集群内唯一
node.id=3
# 集群地址信息
controller.quorum.voters=1@192.168.100.131:9093,2@192.168.100.132:9093,3@192.168.100.133:9093
# 侦听器名称、主机名和代理将向客户端公布的端口.(broker 对外暴露的地址)
advertised.listeners=PLAINTEXT://192.168.100.133:9092,CONTROLLER://192.168.100.133:9093
# kafka数据目录
log.dirs=/data/kafka3. 集群初始化
生成存储目录唯一IDroot@kafka-1:/opt/kafka# bin/kafka-storage.sh random-uuid
1pW25CCcSUmTTXCS8H3qsQ格式化 kafka 存储目录（每个节点都需要执行）root@kafka-1:/opt/kafka# bin/kafka-storage.sh format -t 1pW25CCcSUmTTXCS8H3qsQ -c /opt/kafka/con4. 集群启动
systemd管理
# 注册到systemd管理, 此处以第一个节点为例，其他类似cat > /etc/systemd/system/kafka.service << EOF[Unit]
Description=Apache Kafka server (KRaft mode)
Documentation=https://kafka.apache.org/documentation/
After=network.target[Service]
Type=simple
User=root
Group=root
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure# 设置 JMX 配置（可选）
Environment=KAFKA_OPTS="-Djava.rmi.server.hostname=192.168.100.131 -Dcom.sun.management.jmxremote.port=9997 -Dcom.sun.management.jmxremote.rmi.port=9997 -Dcom.sun.management.jmxremote. -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false"# 设置 Kafka 日志输出（可选）
StandardOutput=syslog
StandardError=syslog
SyslogIdentifier=kafka[Install]
WantedBy=multi-user.target
EOFsystemctl daemon-reloadsystemctl enable kafka --now手动管理
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.propertiestail -f /opt/kafka/logs/server.log# 检查节点状态bin/kafka-broker-api-versions.sh --bootstrap-server 127.0.0.1:9092bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --listkafka-ui可视化工具
https://github.com/provectus/kafka-ui/releases
1. 下载jar包部署
wget https://github.com/provectus/kafka-ui/releases/download/v0.7.2/kafka-ui-api-v0.7.2.jar2. 编写配置文件
# 配置文件
cat > config.yml << EOF
kafka:clusters:-name: kafka-clusterbootstrapServers: http://192.168.100.131:9092,http://192.168.100.132:9092,http://192.168.100.133:9092metrics:port: 9997type: JMX
EOF
参考样例配置：https://github.com/provectus/kafka-ui/blob/master/documentation/compose/kafka-ui.yaml。
样例文件为环境变量导入配置，如果需要转为配置文件，可使用此工具：https://env.simplestep.ca/进行转换。
3. 启动服务
nohup java -Dspring.config.additional-location=/opt/kafka-ui/config.yml -jar /opt/kafka-ui/kafka-ui-api-v0.7.2.jar &# 推荐使用systemd管理cat > /etc/systemd/system/kafka-ui.service << EOF
[Unit]
Description=Kafka UI Service
After=network.target[Service]
ExecStart=/usr/lib/jvm/java-21-openjdk-amd64/bin/java -Dspring.config.additional-location=/opt/kafka-ui/config.yml -jar /opt/kafka-ui/kafka-ui-api-v0.7.2.jar
User=root
Group=root
WorkingDirectory=/opt/kafka-ui
Restart=always
Environment=JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64 # 根据实际环境设置 JAVA_HOME
Environment=PATH=$PATH:/usr/lib/jvm/java-21-openjdk-amd64/bin[Install]
WantedBy=multi-user.target
EOFsystemctl enable kafka-ui --now基本使用
# 检查节点情况bin/kafka-broker-api-versions.sh --bootstrap-server 127.0.0.1:9092# 创建topicbin/kafka-topics.sh --create --topic test --partitions 3 --replication-factor 2 --bootstrap-server 127.0.0.1:9092# 检查topicbin/kafka-topics.sh --describe --bootstrap-server 127.0.0.1:9092# 生产者发送消息bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic test# 消费者消费消息bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test# 删除topicbin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --delete --topic test
