我们知道kafka的服务注册与发现是由zookeeper来维护，那zookeeper中维护kafka哪些信息，是怎么实现的呢？
1.zookeeper简介
首先看看zookeeper简介，简介参考：https://blog.csdn.net/java_66666/article/details/81015302
zookeeper是一个分布式文件系统，可以像普通文件系统一样增删目录，每个目录节点叫做znode；znode可以存储、修改数据。另外zookeeper的znode是可以被监听的，当节点发送变化时，能够发送消息通知监听者。
zookeeper非常强大，可以用来实现统一内容管理（如统一配置文件）、集群管理（如kafka集群）、命名服务管理（存储服务ip与域名映射）等。
节点包含四种类型：
1）持久化节点--客户端断开后节点依旧存在
2）持久化顺序节点--客户端断开后节点依旧存在，且进行了顺序编号
3）临时节点--客户端断开后节点将删除
4）临时顺序节点--客户端断开后节点将删除，节点进行了顺序编号
使用前面安装的zookeeper进行测试：Mac kafka安装
打开zookeeper客户端，执行zkCli：
mat@matthew-MacBook-Pro bin % zkCli
Connecting to localhost:2181
Welcome to ZooKeeper!
JLine support is enabled
WATCHER::
WatchedEvent state:SyncConnected type:None path:null
[zk: localhost:2181(CONNECTED) 0]
创建持久化节点mx，并赋值data_0：
create /mx data_0
获取节点数据：
[zk: localhost:2181(CONNECTED) 11] get /mx
data_0
2.kafka broker在zookeeper中注册
kafka的broker在启动后就会在/broker/ids下创建一个临时节点，该broker节点为逻辑节点，与具体服务器没有强关联性。该节点为临时节点，broker客户端关闭后将删除。
zkCli中查看本地broker节点：
[zk: localhost:2181(CONNECTED) 3] ls /brokers/ids
[0]
获取brokder信息：
[zk: localhost:2181(CONNECTED) 4] get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://localhost:9092"],"jmx_port":-1,"port":9092,"host":"localhost","version":4,"timestamp":"1605148291772"}
listener_security_protocol_map:定义传输协议
endpoints：定义broker对应的传输协议
jmx_port：JMX监控接口，设置后可获得kafka JMX协议的监控指标
port：brokder端口
host：brokder地址
version：broker版本号
timestamp：brokder启动时间
添加一个本地broker实例
进入kafka配置目录/usr/local/etc/kafka/，拷贝配置文件：
cp server.properties server.properties2
编辑 server.properties2，修改broker.id、端口号、文件存储路径：
broker.id=1
port=9093
log.dirs=/usr/local/var/lib/kafka-logs2
保存，并启动该配置的broker：
kafka-server-start server.properties2
此时再查看/brokers/ids节点下数据：
[zk: localhost:2181(CONNECTED) 5] ls /brokers/ids
[0, 1]
新的broker已经注册上了，查看新broker信息：
[zk: localhost:2181(CONNECTED) 6] get /brokers/ids/1
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://localhost:9093"],"jmx_port":-1,"port":9093,"host":"localhost","version":4,"timestamp":"1605237001847"}
端口为的9093，新的broker完成zookeeper注册。
3.topic在zookeeper中的注册
topic信息都保存在/brokers/topics节点下，创建topic时，会在该节点下以topic名称创建持久化节点，并在该节点下创建partitions节点，partitions下面根据分区数从0开始创建分区节点。
有三个分区的test1节点如下：
[zk: localhost:2181(CONNECTED) 18] ls /brokers/topics/test1/partitions
[0, 1, 2]
各分区节点下的state节点保存分区信息：
[zk: localhost:2181(CONNECTED) 21] get /brokers/topics/test1/partitions/0/state
{"controller_epoch":1,"leader":0,"version":1,"leader_epoch":0,"isr":[0]}
controller_epoch：集群controller的选举次数
leader：leader对应的brokerid
version：版本号默认1
leader_epoch：该partition leader选举次数
isr：该partition同步副本组brokerid
4.consumer在zookeeper中注册
0.9版本以前offset保存在zookeeper consumers节点下，之后就保存在kafka的文件目录下，路径配置项为 log.dir。
5.控制器controller相关信息
旧版本的broker都直接监听zookeeper，造成zk监听过多，分区变动时集群变动过大。后续版本通过选举一个controller来监听维护分区信息。
/controller中信息如下：
{"version":1,"brokerid":0,"timestamp":"1605488793510"}
version：版本号
brokerid：当前controler的broker id
timestamp：选举成功时的时间戳
/controller_epoch节点信息表示当前controller是第几代：
2
总结
到此，我们知道kafka集群broker通过在zookeeper注册节点的形式注册、并发现服务，该节点是临时的，broker退出时节点删除，broker也就退出集群；topic则是注册的持久化节点，以顺序分区编号的方式注册的，并将分区信息存储在其中；consumer0.9版本后由kafka维护，保存在自己的文件目录。
参考：https://kafka.apachecn.org/documentation.html#impl_zktopic
