Kafka集群搭建的两种方式

目录

1. 依赖Zookeeper搭建集群

1. 下载Kafka二进制文件

2. 更改kafka配置

3. 启动Zookeeper集群和Kafka集群

4. 验证集群

1.创建主题

2. 检查主题是否存在

3. 创建生产者生产数据

4. 创建消费者消费数据

5. 检查Zookeeper中Kafka集群的元数据 

2. 不依赖Zookeeper搭建Kafka集群

1.下载Kafka二进制文件

2.更改Kafka配置

3. 生成meta.properties文件

 4. 启动kafka集群

5. 验证集群

1. 创建主题，并验证主题存在

2. 生产者生产数据

3. 消费者消费数据

---

1. 依赖Zookeeper搭建集群

在2.8.0版本以前，Kafka集群的搭建依赖于Zookeeper环境，需要将kafka集群节点ip及主题等Kafka相关的元数据存入Zookeeper服务中。所以搭建Kafka集群前，需要先搭建Zookeeper集群。

zookeeper多主机集群搭建https://blog.csdn.net/dxh9231028/article/details/141052933?spm=1001.2014.3001.5501

1. 下载Kafka二进制文件

搭建完Zookeeper集群后，从Kafka官网下载Kafka二进制文件，这里我们下载最新的3.8.0版本。

然后创建/kafka文件夹，将文件传入该文件夹解压，并将解压后的文件改名为kafka 。

#解压文件
tar -xf ./kafka_2.13-3.8.0.tgz
#更改名字
mv ./kafka_2.13-3.8.0 ./kafka

这里我们搭建三个节点的Kafka集群，所以要再三个主机都进行如上操作。

2. 更改kafka配置

修改kafka/conf/server.properties文件中的如下配置

#集群中kafka节点的唯一id，其他的节点要配置别的id
broker.id=0 #Kafka启动时会向Zookeeper集群进行服务注册，这个配置项记录所有Zookeeper节点的ip和端口，只记录一个localhost也能成功启动，不过为了集群的高可用性，全部配置更好。
zookeeper.connect=ip1:2181,ip2:2181,ip3:2181 #Kafka监听的IP端口，由于现在的主题通常是多网卡环境，所以需要指定ip，早期版本好像可以使用0.0.0.0监听全部的ip，不过这个版本不可以，必须指定明确的ip。
listeners=PLAINTEXT://ip1:9092#kafka日志的存储位置，实际上是Kafka的全部数据，包括Kafka内部存储的消息，以及Kafka集群第一次启动时从Zookeeper中获取的集群元数据。
log.dirs=/path/to/kafka/logs

启动一个简单的集群，Kafka不需要太多的配置，不过实际生产环境中Kafka的部署需要考虑大量因素，需要大量的调整配置，运维难度较大。

3. 启动Zookeeper集群和Kafka集群

如果在Zookeeper集群搭建后已经启动了，这里就不需要在启动了，如果没有启动，不仅可以通过Zookeeper提供的zkServer.sh可执行文件启动，也可以通过Kafka的可执行文件启动。

#启动Zookeeper，如果之前没启动
bin/zookeeper-server-start.sh config/zookeeper.properties#启动kafka
bin/kafka-server-start.sh config/server.properties#后台启动Kafka，不占用当前终端
nohup bin/kafka-server-start.sh config/server.properties > kafka.out 2>&1 &

4. 验证集群

集群启动成功后，我们可以通过创建生产者，生产数据，并在另一个Kafka服务进行消费来验证集群创建是否成功。

1.创建主题

通过调用Kafka的bin目录下的kafka-topic.sh文件创建主题，命令如下。

./bin/kafka-topics.sh --create --topic test-topic --bootstrap-server 192.168.142.20:9092 --replication-factor 3 --partitions 3
#create代表当前执行的创建主题操作
#topic指定创建主题名称
#bootstrap-server指定从哪个kafka服务实例连接集群，可以指定多个，防止某个节点宕机造成指令执行失败。
#replication-factor指定当前主题要复制几份，分给其他Kafka服务实例，当前是三节点集群，我们创建三份，保证每个节点都有该主题数据。
#当复制份数少于集群节点数时会有部分节点没有该主题数据，不过不会影响kafka集群正常消费数据。

创建成功后结果如下图

2. 检查主题是否存在

依旧是调用Kafka-topic.sh文件，检查主题是否存在

./bin/kafka-topics.sh --list --bootstrap-server 192.168.142.21:9092
#list指的是当前是要执行查看主题列表操作
#bootstrap-server指定接入集群的kafka服务地址

这里我们通过第二台主机，连接第二个kafka节点获取主题列表，结果如下

3. 创建生产者生产数据

确定主题存在后，我们再用这台主机，和这个kafka节点服务创建生产者生产数据。

bin/kafka-console-producer.sh --topic test-topic --bootstrap-server 192.168.142.21:9092

 仔细阅读上面的配置解释，这里的配置就不解释了，应该很容易看懂，结果如下

可以看到执行完可执行文件后，会出现特殊的命令行，在这个命令行填写数据并回车，则可以向指定的主题消费数据，这里我生产了三条数据。

4. 创建消费者消费数据

这里我们再用第三台主机和第三个kafka节点服务来消费数据，通过可执行文件kafka-console-consumer.sh执行以下命令

bin/kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server 192.168.142.22:9092

结果如下

可以看到成功消费到生产者生产的数据

5. 检查Zookeeper中Kafka集群的元数据 

通过zkCli.sh执行如下命令连接zookeeper。

./zkCli.sh -server 192.168.142.20

连接后通过调用ls -R / 查看zookeeper中所有的节点信息，内容过长，这里无法截图，直接复制到代码块中，并且删除掉__consumer_offset节点下的所有子节点信息，方便大家查看。

从如下结果可以看到，除了/zookeeper节点外，其他节点均为kafka集群元数据信息，其中/brokers/ids节点由0，1，2三个子节点，分别记录了Kafka集群中每个节点的ip端口等信息，而/brokers/topics下保存了所有主题信息，其中可以看到除了我们创建的test-topic主题外，还有__consumer_offsets主题，这个主题中记录了所有消费者组和其负责的主题分区的对应关系，内容极多，这里为了方便大家观看结果，将其删除，想了解更多可以看我的另一篇文章

Kafka基本概念及消费流程https://blog.csdn.net/dxh9231028/article/details/141270920?spm=1001.2014.3001.5501

/
/admin
/brokers
/cluster
/config
/consumers
/controller
/controller_epoch
/feature
/isr_change_notification
/latest_producer_id_block
/log_dir_event_notification
/zookeeper
/admin/delete_topics
/brokers/ids
/brokers/seqid
/brokers/topics
/brokers/ids/0
/brokers/ids/1
/brokers/ids/2
/brokers/topics/__consumer_offsets
/brokers/topics/test-topic
/brokers/topics/test-topic/partitions
/brokers/topics/test-topic/partitions/0
/brokers/topics/test-topic/partitions/1
/brokers/topics/test-topic/partitions/2
/brokers/topics/test-topic/partitions/0/state
/brokers/topics/test-topic/partitions/1/state
/brokers/topics/test-topic/partitions/2/state
/cluster/id
/config/brokers
/config/changes
/config/clients
/config/ips
/config/topics
/config/users
/config/topics/__consumer_offsets
/config/topics/test-topic
/zookeeper/config
/zookeeper/quota

2. 不依赖Zookeeper搭建Kafka集群

Kafka 从版本 2.8.0 开始，引入了新的 KRaft 模式，这使得 Kafka 可以在不依赖 ZooKeeper 的情况下部署集群。在传统的 Kafka 架构中，ZooKeeper 用于管理和协调 Kafka 集群中的元数据，例如主题、分区、副本状态等。KRaft 模式则将这些管理功能集成到 Kafka 自身的控制器节点中，消除了对 ZooKeeper 的依赖。

1.下载Kafka二进制文件

第一步仍然是下载解压文件，和之前一样

2.更改Kafka配置

Kafka默认配置是基于Zookeeper的，所以我们首先要删除Zookeeper相关的配置，通过vim命令中/zookeeper查找内容，注释或删除掉相关配置，然后删除掉broker.id，排除了zookeeper模式相关配置的影响后在进行如下配置（不删除可能也不影响启动，不过最好删除，防止其他不可预料的行为）。

#指定 Kafka 节点的角色，可以是broker、controller，或两者兼具。
process.roles=broker,controller#当前节点的唯一id
node.id=1#kafka集群中担任controller角色的kafka服务地址
controller.quorum.voters=1@192.168.142.30:9093,2@192.168.142.31:9093,3@192.168.142.32:9093#当前的节点中由controller角色，则需要配置一个name标识，必须大写
controller.listener.names=CONTROLLER#为上面设置的controller标识设置一个ip地址，这个地址也是controller.quorum.voters中配置的地址
listeners=PLAINTEXT://192.168.142.30:9092,CONTROLLER://192.168.142.30:9093#存储之前存储在Zookeeper中的元数据内容
metadata.log.dir=/var/lib/kafka/meta#配置初始化broker角色的kafka实例的超时时间，默认是30s，这里配置高一点，下面会说原因
initial.broker.registration.timeout.ms=240000

在传统的zookeeper模式下，zookeeper担任着选举leader及各种集群元数据管理的工作，而在新版本的kraft模式下，controller角色的kafka服务实例接替了这一工作，代替zookeeper处理Leader选举以及元数据管理的工作，并将元数据存入metadata.log.dir配置的路径中。

3. 生成meta.properties文件

在传统的zookeeper模式下，kafka集群信息保存在zookeeper中，也就是说只要Kafka实例·将自己的信息注册在zookeeper中，它就是集群的一部分，区分不同集群的根据是不同的zookeeper集群。

而在kraft模式下，kafka并没有像redis或zookeeper那样，将集群中所有节点的ip地址配置到配置文件中，而是通过kafka-storage.sh可执行文件生成meta.properties文件，命令如下

bin/kafka-storage.sh format --config conf/server.properties --cluster-id 123
#config选项指定配置文件路径，获取配置文件中的nodeid
#cluster-id指定一个集群id

实际生产过程中，使用Kafka提供的bin/kafka-storage.sh random-uuid生成这两个id，bin/kafka-storage.sh random-uuid可以生成一个唯一id，用于nodeid或clusterid，这里我们定义简单的nodeid和clusterid用于测试Kafka集群搭建过程。

每个Kafka实例都要执行这个命令，生成一个meta.properties文件，文件内容如下：

#
#Sat Aug 17 02:55:59 PDT 2024
cluster.id=123
version=1
directory.id=AABGMd7o4oHy2t-qzVzhcQ
node.id=1

其中cluster.id就是集群标识，一个集群中的所有Kafka节点实例的meta.properties文件中的cluster.id必须相同，并且node.id必须不同。

Kafka没有像其他中间件那样将节点ip写在配置中，可能的原因是想增加可扩展性。如果将所有ip写在配置中，意味着如果我想新增一个节点则需要修改全部kafka实例的配置文件，而如果像Kafka这样，利用一个文件中的cluster-id选项区分是否是一个集群，这样当想要新增一个Kafka实例时，只需要让新增的Kafka实例生成的meta.properties中的集群id相同即可。

 4. 启动kafka集群

启动kafka集群命令和传统模式一样，都是通过kafka-server-start.sh可执行文件，启动Kafka服务。

bin/kafka-server-start.sh config/server.properties#后台启动
nohup bin/kafka-server-start.sh /path/to/kraft-server.properties > kafka.out 2>&1 &

不同的时，在传统模式下，集群的元数据的处理是靠已经启动的zookeeper，这是各个Kafka实例可以随时启动。而在kraft模式下，集群的元数据处理依赖于controller节点，所以在启动broker角色的Kafka实例之前，必须将controller全部启动完成，然后broker才会正常启动。

但如果节点既是controller，又是broker呢？那么此时这个节点的controller角色将会正常启动，不过由于controller没有全部启动，broker角色无法初始化成功，在这种情况下，该Kafka实例会不断地循环初始化，直到controller节点全部启动后，broker角色才会初始化成功。一旦brocker角色循环初始化的时间超过了超时时间（默认是30s），Kafka服务就会启动失败。

所以在配置项中我们新增一项initial.broker.registration.timeout.ms=240000，将broker初始化的超时时间设置为240s，让我们由充足的时间启动全部的controller角色的Kafka实例。

5. 验证集群

验证集群和之前一样，通过在不同主机和Kafka实例上操作生产者，消费者在主题中生产或消费信息

1. 创建主题，并验证主题存在

2. 生产者生产数据

3. 消费者消费数据

与之前不同的是，kraft模式下，集群相关元数据存储controller角色，在 metadata.log.dir 配置项配置的文件夹下，内容如下

具体内容是什么我也不知道，也没必要深究。