当前位置：首页 > news >正文

Kakfa集群部署及主题创建

news 2025/7/20 10:55:05

前言

上篇文章中Kafka的基本使用-CSDN博客主要介绍了Kafka在Windows环境下的安装和基本使用。这篇文章主要介绍Kafka在集群环境中部署相关概念。

生产环境都是采用linux系统搭建服务器集群，但是重点是在于学习kafka的基础概念和核心组件，所以这里搭建一个简单易用的windows集群方便大家的学习和练习。

Kafka集群

Kafka集群部署

1. 在磁盘根目录创建文件夹cluster，将下载的kafka_2.13-3.8.0安装包解压在cluster目录下。

2. 将kafka_2.13-3.8.0重命名为kafka-node-1,并复制3份，命名为kafka-node-2，kafka-node-3。

3. 再复制一份，重命名为kafka-zookeeper。

修改配置文件

1. 修改config/zookeeper.properties文件

2. 分别修改kafka-node-1、kafka-node-2、kafka-node-3中config/server.properties配置文件。

分别修改三个Kafka的borker.id为1,2,3。以及三个Kafka节点的端口9091、9092、9093。并修改三个Kafka中数据文件路径和Zookeeper软件连接地址。例如：

broker.id=1

listeners=PLAINTEXT://:9091

log.dirs=E:/cluster/kafka-node-1/data

zookeeper.connect=localhost:2181/kafka

封装启动脚本

因为Kafka启动前，必须先启动ZooKeeper，并且Kafka集群中有多个节点需要启动，所以启动过程比较繁琐，这里我们将启动的指令进行封装。

在kafka-zookeeper文件夹下创建zk.cmd批处理文件

在zk.cmd中添加如下命令：

call bin/windows/zookeeper-server-start.bat config/zookeeper.properties

在kafka-node-1，kafka-node-2，kafka-node-3文件夹下分别创建kfk.cmd批处理文件

在kfk.cmd中添加如下命令：

call bin/windows/kafka-server-start.bat config/server.properties

在cluster文件夹下创建cluster.cmd批处理文件，用于启动kafka集群

在cluster.cmd文件中添加内容

cd kafka-zookeeper

start zk.cmd

ping 127.0.0.1 -n 10 >nul

cd ../kafka-node-1

start kfk.cmd

cd ../kafka-node-2

start kfk.cmd

cd ../kafka-node-3

start kfk.cmd

在cluster文件夹下创建cluster-clear.cmd批处理文件，用于清理和重置kafka数据

在cluster-clear.cmd中添加如下命令：

cd kafka-zookeeper

rd /s /q data

cd ../kafka-node-1

rd /s /q data

cd ../kafka-node-2

rd /s /q data

cd ../kafka-node-3

rd /s /q data

双击执行cluster.cmd文件，启动Kafka集群

集群启动命令后，会打开多个黑窗口，每一个窗口都是一个kafka服务，请不要关闭，一旦关闭，对应的kafka服务就停止了。如果启动过程报错，主要是因为zookeeper和kafka的同步问题，请先执行cluster-clear.cmd文件，再执行cluster.cmd文件即可。

Kafka集群启动

Controller选举

第一次启动Kafka集群时，会同时启动多个Broker，每个Broker都会连接到Zookpeeper，并尝试创建一个/Controller节点。

因为Zookpeeper中同一个节点不允许重复创建，所以当有多个Broker时，最终只有一个Broker能创建成功，创建成功的这个Broker就会成为Kafka的集群控制器节点。用来管理整个Kafka集群

没有选举成功的Slava节点就会创建一个监听器，用于监听/controller节点的变化

当controller节点出现故障或者挂掉了，那么此时对应的Zookpeeper连接就会中断，Zookpeeper中的/controller节点就会自动删除掉，而其他的Slave节点因为添加了监听器，当监听到/controller节点被删除后，就会马上向Zookpeeper发出创建/controller节点的请求，一旦创建成功，那么该Broker就会成为新的Controller了。

集群中主题的创建

创建主题

创建主题Topic的方式有很多种：命令行，工具，客户端API，自动创建。在server.properties文件中配置参数auto.create.topics.enable=true时，如果访问的主题不存在，那么Kafka就会自动创建主题。

我们采用javaAPI的方式创建主题

public class AdminTopicTest {public static void main(String[] args) {Map<String,Object> map = new HashMap<>();map.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 管理员对象Admin admin = Admin.create(map);
//        构建主题时需要传递三个参数：主题名称，分区数，副本数String topicName = "topic1";int partitionCount = 1;short replicationCount = 1;NewTopic topic1 = new NewTopic(topicName, partitionCount, replicationCount);String topicName2 = "topic2";int partitionCount1 = 2;short replicationCount1 = 2;NewTopic topic2 = new NewTopic(topicName2, partitionCount1, replicationCount1);// 创建主题CreateTopicsResult result = admin.createTopics(Arrays.asList(topic1, topic2));
//        自定义副本分配策略Map<Integer, List<Integer>> configs = new HashMap<>();configs.put(0, Arrays.asList(1,2));  // 分区0的副本分配策略 1号服务器为Leader分区，2号服务器为Follower分区configs.put(1, Arrays.asList(2,3));configs.put(2, Arrays.asList(1,3));NewTopic test = new NewTopic("test",configs);admin.createTopics(Arrays.asList(test));// 关闭资源admin.close();}
}

创建主题原理：集群中创建主题的操作是由controller节点完成的，我们连接的不一定是controller节点，所有当我们向我们连接的broker发送创建主题的请求时，首先连接的该broker会获取到controller节点的位置，然后返回给Admin管理员对象，然后再向controller节点发送创建主题的请求，然后controller节点创建主题成功后，其他broker在监听/brokers/topics时会发现创建的主题，就会将创建的主题同步到各个broker上。

下篇文章将主要介绍Kakfa在集群环境中数据的生产和发送相关操作。

查看全文

http://www.dtcms.com/a/288170.html