Kafka集群的搭建与批量启停方式

更新时间：2025年11月12日 09:51:59 作者：薛伟同学

Kafka是一个分布式的基于发布/订阅模式的消息队列,适用于大数据实时处理,本文详细介绍了如何在CentOS 7上搭建Kafka集群,包括服务器配置、配置文件修改、服务启动和关闭脚本编写

什么是Kafka？

Kafka是一个分布式的基于发布/订阅模式的消息队列（MessageQueue），主要应用于大数据实时处理领域。

最新官方给Kafka的定义是一个开源的分布式事件流平台（Event Streaming Platform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。

Kafka集群的搭建

部署服务器发行版本为CentOS7 ✖️ 三台（已安装好Java环境），三台服务器的IP为：

192.168.182.4
192.168.182.5
192.168.182.6

部署Kafka版本为kafka_2.12-3.1.0

下载安装包、上传服务器

https://kafka.apache.org/downloads

将下载好的安装包上传至每台服务器的/opt/module/目录下后解压重命名为kafka。

修改server.properties

分别进入三台服务器kafka的配置文件目录config，修改服务配置文件server.properties。

将broker.id分别修改为0、1、2（不可重复）
添加listeners=PLAINTEXT://192.168.182.4:9092配置，对应好每个服务器的IP地址
修改log.dirs=/opt/module/kafka/datas
修改zookeeper.connect=192.168.182.4:2181,192.168.182.5:2181,192.168.182.6:2181

主要修改部分

...
broker.id=0
listeners=PLAINTEXT://192.168.182.4:9092
log.dirs=/opt/module/kafka/datas
zookeeper.connect=192.168.182.4:2181,192.168.182.5:2181,192.168.182.6:2181
...

...
broker.id=1
listeners=PLAINTEXT://192.168.182.5:9092
log.dirs=/opt/module/kafka/datas
zookeeper.connect=192.168.182.4:2181,192.168.182.5:2181,192.168.182.6:2181
...

...
broker.id=2
listeners=PLAINTEXT://192.168.182.6:9092
log.dirs=/opt/module/kafka/datas
zookeeper.connect=192.168.182.4:2181,192.168.182.5:2181,192.168.182.6:2181
...

修改zookeeper.properties

分别进入三台服务器kafka的配置文件目录config，修改服务配置文件修改zookeeper.properties。

修改dataDir=/opt/module/kafka/zk-datas，后需要手动创建该目录，并创建myid文件，三台服务器的myid不可重复，分别为0，1，2

添加节点的配置：

server.0=192.168.182.4:2888:3888
server.1=192.168.182.5:2888:3888
server.2=192.168.182.6:2888:3888

添加配置：

initLimit=10
syncLimit=5
tickTime=2000

三台服务器保持一致，像下面这样

dataDir=/opt/module/kafka/zk-datas
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
initLimit=10
syncLimit=5
tickTime=2000
maxClientCnxns=0
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080
server.0=192.168.182.4:2888:3888
server.1=192.168.182.5:2888:3888
server.2=192.168.182.6:2888:3888

修改bin/kafka-server-stop.sh

官方的停止服务的脚本不太好用，将三台服务器的kafka-server-stop修改成下面的内容

#!/bin/sh

PIDS=$(jps -lm | grep -i "kafka.Kafka" | awk '{print $1}')

if [ -z "$PIDS" ]; then
  echo "No kafka server to stop"
  exit 1
else
  kill -s TERM $PIDS
fi

注意：当使用ssh来远程执行此脚本的时候，可能会报jps命令找不到，但是在服务器本身上执行是没有问题，原因是你服务器的jps命令不在Path中，加上去就行，或者使用jps的绝对路径。

启动测试

一定要先启动zk再启动kafka，关闭的时候先关闭kafka，再关闭zk。

启动Zookerper

三台服务器上分别执行下面语句启动Zookerper

# 前台启动
/opt/module/kafka/bin/zookeeper-server-start.sh /opt/module/kafka/config/zookeeper.properties
# 后台启动
/opt/module/kafka/bin/zookeeper-server-start.sh -daemon /opt/module/kafka/config/zookeeper.properties

jps -lm可以看到zk已经启动成功。

启动Kafka

三台服务器上分别执行下面语句启动Kafka

# 前台启动
/opt/module/kafka/bin/kafka-server-start.sh /opt/module/kafka/config/server.properties
# 后台启动
/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties

jps -lm可以看到kafka已经启动成功。

关闭Kafka

三台服务器上分别执行下面语句关闭Kafka

/opt/module/kafka/bin/kafka-server-stop.sh

jps -lm可以看到kafka已经关闭成功。

关闭Zookerper

三台服务器上分别执行下面语句关闭Zookerper

/opt/module/kafka/bin/kafka-server-stop.sh

jps -lm可以看到zookerper已经关闭成功。

编写集群启停脚本

刚才我们是分别在每台服务器上执行启动、停止的命令，这种方式不够方便。所以我们可以通过ssh远程命令来远程在集群中的一台服务器上远程执行另外两台服务器的命令，前提是需要配置好ssh免密登陆，前面的文章与有介绍。

Zookerper的集群启停脚本

创建zk.sh文件，输入以下内容：

#!/bin/bash

# description: zk集群启动脚本
# author: 薛伟

case $1 in
"start"){
        for i in 192.168.182.4 192.168.182.5 192.168.182.6
        do 
                echo " --------启动 $i Kafka-------"
                ssh $i "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"
        done
};;
"stop"){
        for i in 192.168.182.4 192.168.182.5 192.168.182.6
        do 
                echo " --------停止 $i Kafka-------"
                ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh"
        done
};;
esac

赋予运行权限chmod +x zk.sh

启动集群：./zk.sh start
关闭集群：./zk.sh stop

Kafka的集群启停脚本

创建kafka.sh文件，输入以下内容：

#!/bin/bash

# description: zk集群启动脚本
# author: 薛伟

case $1 in
"start"){
        for i in 192.168.182.4 192.168.182.5 192.168.182.6
        do 
                echo " --------启动 $i Kafka-------"
                ssh $i "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"
        done
};;
"stop"){
        for i in 192.168.182.4 192.168.182.5 192.168.182.6
        do 
                echo " --------停止 $i Kafka-------"
                ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh"
        done
};;
esac

赋予运行权限chmod +x kafka.sh

启动集群：./kafka.sh start
关闭集群：./kafka.sh stop

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

springboot IDEA启动两个端口服务nginx负载过程
这篇文章主要介绍了springboot IDEA启动两个端口服务nginx负载过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2025-03-03
SpringBoot 快速实现分库分表的2种方式
本文将为您介绍 ShardingSphere 的一些基础特性和架构组成，以及在 Springboot 环境下通过JAVA编码和Yml配置两种方式快速实现分库分表功能，感兴趣的朋友跟随小编一起看看吧
2023-06-06
Java Socket编程实例（五）- NIO UDP实践
这篇文章主要讲解Java Socket编程中NIO UDP的实例，希望能给大家做一个参考。
2016-06-06
java巧用@Convert实现表字段自动转entity
本文主要介绍了java巧用@Convert实现表字段自动转entity,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2024-07-07
解决springboot 2.x 里面访问静态资源的坑
这篇文章主要介绍了解决springboot 2.x 里面访问静态资源的坑，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-08-08
SpringCloud基于Feign的可编程式接口调用实现
本文主要介绍了SpringCloud基于Feign的可编程式接口调用实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2022-04-04
详解SpringBoot中使用RabbitMQ的RPC功能
这篇文章主要介绍了详解SpringBoot中使用RabbitMQ的RPC功能,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2021-11-11
在Java中Collection的一些常用方法总结
今天给大家带来的知识是关于Java的,文章围绕着Java中Collection的一些常用方法展开,文中有非常详细的介绍及代码示例,需要的朋友可以参考下
2021-06-06
IDEA的Mybatis Generator驼峰配置问题
这篇文章主要介绍了IDEA的Mybatis Generator驼峰配置问题,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2020-11-11
SpringBoot激活profiles的几种方式
多环境是最常见的配置隔离方式之一，可以根据不同的运行环境提供不同的配置信息来应对不同的业务场景，这篇文章主要介绍了SpringBoot激活profiles的几种方式，需要的朋友可以参考下
2019-10-10