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

总结

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

您可能感兴趣的文章:

Java实现新建有返回值的线程的示例详解
本文主要介绍了一个Java多线程的例题，题目是：使用ThreadLocal管理一号和二号线程，分别存入100元，在三号线程中使用利用一号和二号的计算结果来算出账户的实际金额。感兴趣的可以了解一下
2022-09-09
SpringBoot详解如何进行整合Druid数据源
Druid是阿里开发的一款开源的数据源，被很多人认为是Java语言中最好的数据库连接池，本文主要介绍了SpringBoot整合Druid数据源的方法实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2022-06-06
利用Kafka动态调整topic分区partition
这篇文章主要介绍了利用Kafka动态调整topic分区partition问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-12-12
SpringBoot集成Mybatis过程步骤图解
这篇文章主要介绍了SpringBoot集成Mybatis过程步骤图解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-07-07
java中使用logger打印日志有哪些坑
在Java中使用日志记录器（Logger）打印日志时,确实存在一些常见的坑需要注意,本文就来介绍一下几种方法,具有一定的参考价值,感兴趣的可以了解一下
2024-05-05
springcloud项目占用内存好几个G导致服务器崩溃的问题
这篇文章主要介绍了springcloud项目占用内存好几个G导致服务器崩溃的问题,本文给大家分享解决方案供大家参考，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2020-10-10
Spring循环依赖的三种方式(推荐)
本篇文章主要介绍了Spring循环依赖的三种方式，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2018-01-01
SpringCloud整合Netty集群实现WebSocket的示例代码
文章主要介绍了SpringCloud整合Netty集群实现WebSocket的相关内容,包括服务注册和发现中心的配置,如使用Nacos、CommandLineRunner启动Netty服务等,还介绍了通过Redis实现消息发布订阅的机制,需要的朋友可以参考下
2024-11-11
Linux下用java -jar运行可执行jar包的方法教程
这篇文章主要给大家介绍了在Linux下用java -jar运行可执行jar包的方法教程，文中介绍的非常详细，相信对大家的工作或者学习具有一定的参考学习价值，需要的朋友们下面来一起看看吧。
2017-05-05
Java建造者设计模式详解
这篇文章主要为大家详细介绍了Java建造者设计模式，对建造者设计模式进行分析理解，感兴趣的小伙伴们可以参考一下
2016-02-02