系列文章目录

Kafka第一章：环境搭建
Kafka第二章：生产者案例
Kafka第三章：新旧节点更替

---

---

前言

这次博客，我们学习一下Kafka新旧节点的更替，比如一些硬件老化或者更新换代，如何添加新节点，如何去除旧节点

---

一、创建新节点

1.克隆节点

我们用hadoop104直接克隆一个hadoop105，具体方法可以看我另一个博客。hadoop环境搭建，克隆完成之后修改IP和用户名，然后重启虚拟机即可。

2.修改Kafka配置

修改 haodoop105 中 kafka 的 broker.id 为 3。

vim config/server.properties

删除 hadoop105 中 kafka 下的 datas 和 logs

rm -rf datas/ logs/

二、添加新节点

1.启动集群

2.启动105的Kafka

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

3.创建一个要均衡的主题

这个操作在102上进行

vim topics-to-move.json

{"topics": [{"topic": "first"}],"version": 1
}

4.生成一个负载均衡的计划

注意–broker-list指定四个几点

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate

这是修改前后，三个分区内容的存储节点的ID，可以明显看到新节点已经加入。

5.创建副本存储计划

 vim increase-replication-factor.json

#这里是刚才生成的新分区信息，复制你自己的
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,2,3],"log_dirs":["any","any","any"]}]}

6.执行副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

我们可以查看查看一下topic信息。

三、去除旧节点

理论上直接把就节点直接干掉就行，但是不建议，就好像你电脑关机，会使用关机按钮，而不是直接拽电源。
去除方法和添加方法类似,我们只需要修改之前的执行计划，其他几乎一样。

1.创建执行计划

我们将这里的–broker-list 换成去除节点后的ID即可。

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate

2.修改副本存储计划

换成新的

 vim increase-replication-factor.json

{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,0,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,2,1],"log_dirs":["any","any","any"]}]}

3.执行副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

查看存储节点

已经去除了目标节点。

---

总结

这一次，Kafka新旧节点更换，暂时就到这里。