启用集群

配置集群时关键就是提前梳理好需要的网络资源，完成对应server.properties文件的配置。在执行前先把这些梳理好，可以方便后面的配置，也不容易出错，其他操作跟单机Kafka版都是一样的。
一般需要提前规划各节点所在服务器、端口、Kafka安装路径、数据存放位置、配置文件位置、ssl证书位置（需要配置的ssl的话才需要）。对于测试，如果所有节点都在同一台服务器上，直接使用同一个Kafka搭配多个不同配置文件启动多个实例就行。

资源规划

下面是通过一台服务器的多个端口仿照多台服务器，配置3个节点的集群，大概需要以下资源：

服务器地址	broker端口	controller端口	安装位置	配置文件	数据	证书位置
192.168.1.1	9092	9095	/home/kafka-cluster/broker1	/home/kafka-cluster/broker1/config	/data/kafka/broker1	/usr/ca/ssl
192.168.1.1	9093	9096	/home/kafka-cluster/broker2	/home/kafka-cluster/broker2/config	/data/kafka/broker2	/usr/ca/ssl2
192.168.1.1	9094	9097	/home/kafka-cluster/broker3	/home/kafka-cluster/broker3/config	/data/kafka/broker3	/usr/ca/ssl3

如果是在多台服务器上配置，可以直接配置好一台，复制server.properties或者整个kafka到其他服务器，这样可以减少很多重复性工作。只需要保证把证书生成到相同的路径下，就不用做重复其他工作了。

准备证书

证书生成可以参考上一篇单机Kafka配置ssl并在springboot使用
生成的各节点证书如下：

创建相关文件夹

Kafka及配置文件位置：

各节点数据位置：

配置文件

broker1配置文件：

#SSL配置部分
listeners=SSL://:9092
#下面的localhost需要改成ip，否则只有自己能连上
advertised.listeners=SSL://localhost:9092
security.inter.broker.protocol=SSL
ssl.endpoint.identification.algorithm=
ssl.keystore.location=/usr/ca/ssl/keystore.jks
ssl.keystore.password=123456
ssl.key.password=123456
ssl.truststore.location=/usr/ca/ssl/truststore.jks
ssl.truststore.password=123456
#这里配置成双向认证
ssl.client.auth=required
# 不验证客户端证书
#ssl.client.auth=none

# 在集群中的编号，只要保持唯一就行，
broker.id=1

#Kraft模式的集群关键配置就是这一句，每个节点多配置了一个controller端口
controller.quorum.voters=1@kafka1:9095,2@kafka1:9096,3@kafka1:9097

log.dirs=/data/kafka/broker1
#后面的配置都是默认的，可以不用管
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
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
#虽然Kraft模式的集群不需要zookeeper，但是这个配置是必须要有的，否则会报错
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

broker2配置文件:

listeners=SSL://:9093
#下面的localhost需要改成ip，否则只有自己能连上
advertised.listeners=SSL://localhost:9093
security.inter.broker.protocol=SSL
ssl.endpoint.identification.algorithm=
ssl.keystore.location=/usr/ca/ssl2/keystore.jks
ssl.keystore.password=123456
ssl.key.password=123456
ssl.truststore.location=/usr/ca/ssl/truststore.jks
ssl.truststore.password=123456
#这里配置成双向认证
ssl.client.auth=required
# 不验证客户端证书
#ssl.client.auth=none  

broker.id=2

controller.quorum.voters=1@kafka1:9095,2@kafka1:9096,3@kafka1:9097
log.dirs=/data/kafka/broker2

#后面的配置都是默认的，可以不用管
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
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
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

broker3配置文件：

listeners=SSL://:9094
#下面的localhost需要改成ip，否则只有自己能连上
advertised.listeners=SSL://localhost:9094
security.inter.broker.protocol=SSL
ssl.endpoint.identification.algorithm=
ssl.keystore.location=/usr/ca/ssl3/keystore.jks
ssl.keystore.password=123456
ssl.key.password=123456
ssl.truststore.location=/usr/ca/ssl/truststore.jks
ssl.truststore.password=123456
#这里配置成双向认证
ssl.client.auth=required
# 不验证客户端证书
#ssl.client.auth=none 

broker.id=3

controller.quorum.voters=1@kafka1:9095,2@kafka1:9096,3@kafka1:9097
log.dirs=/data/kafka/broker3

#后面的配置都是默认的，可以不用管
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
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
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

集群关键配置：

broker.id=3
controller.quorum.voters=1@kafka1:9095,2@kafka1:9096,3@kafka1:9097 

其他的配置项单机也要配置，集群只是保证broker id的唯一性且增加controller配置，controller配置中的kafka1为对应服务器hosts的主机名。

启动各Kafka节点

cd到对应Kafka路径下（比如broker1就cd /home/kafka-cluster/broker1）执行以下命令或者使用绝对路径执行，

./bin/kafka-server-start.sh   -daemon  config/server.properties

故障转移测试

故障前的集群状态：

原leader故障后从broker1转到broker2

恢复原来的leader节点之后，会自动切回去

spring boot集成

其他配置跟单机Kafka配置ssl并在springboot使用是一样的，只需要把bootstrap-servers改成多个节点的情况就行，其中kafka是项目服务器hosts配置的主机名

bootstrap-servers: kafka:9092,kafka:9093,kafka:9094