下载地址
https://archive.apache.org/dist/kafka
上传并解压
tar -zxvf kafka_2.13-3.6.1.tgz
建立软连接
ln -s /usr/local/kafka_2.13-3.6.1 kafka
创建数据目录
mkdir -p /home/data/kafka
修改配置
多台服务器都要修改
vi /usr/local/kafka/config/kraft/server.properties
node.id=1 # 节点ID，每个节点的值要不同
controller.quorum.voters=1@pro30:9093,2@pro31:9093,3@pro32:9093 # Controller节点配置，用于管理状态的节点（替换Zookeeper作用）
advertised.listeners=PLAINTEXT://192.168.8.198:9092 # 使用IP端口，每个节点填写自己节点的IP，角色代理向外暴露的IP+端口
log.dirs=/home/data/kafka # 数据存储位置
log.retention.hours=168 # 消息默认一周时间
初始化集群
在其中一台服务器上执行下面命令生成一个uuid
KAFKA_CLUSTER_ID="$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)"
用该uuid格式化kafka存储目录，三台服务器都要执行以下命令
/usr/local/kafka/bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /usr/local/kafka/config/kraft/server.properties
启动服务(后台启动)
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/kraft/server.properties
验证
查看日志
tail -n 200 /usr/local/kafka/logs/kafkaServer.out
查看进程
ps -aux | grep kafka
配置服务管理脚本
cat > /usr/lib/systemd/system/kafka.service << EOF
[Unit]
Description=Apache Kafka server (broker)
After=network.target[Service]
Type=forking
User=root
Group=root
Environment="JAVA_HOME=/usr/local/jdk1.8.0_202"
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/kraft/server.properties
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh
Restart=on-failure[Install]
WantedBy=multi-user.target
EOF
# 加载配置
systemctl daemon-reload
# 开机启动
systemctl enable kafka
# 查看服务状态
systemctl status kafka
)
