Zookeeper在kafka作用(zookeeper与kafka的关系)

Kafaka入门（1）- Kafka简介和安装与启动（mac）

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。kafka 是一个高性能的消息队列，也是一个分布式流处理平台。

kafka中文网

kafka官网

Producer ：Producer即生产者，消息的产生者，是消息的入口。

kafka cluster ：

Broker ：Broker是kafka实例，每个服务器上有一个或多个kafka的实例，姑且认为每个broker对应一台服务器。一个集群由多个broker组成，集群内的broker都有一个不重复的编号，如图中的broker-0、broker-1等……

Topic ：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。

Partition ：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。 同一个topic在不同的分区的数据是不重复的 ，partition的表现形式就是一个一个的文件夹！

Replication : 每一个分区都有多个副本 ，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。

Message ：每一条发送的消息主体。

Consumer ：消费者，即消息的消费方，是消息的出口。

Consumer Group ：将多个消费组成一个消费者组。在kafka的设计中 同一个分区的数据只能被同一消费者组中的某一个消费者消费 。Partition 的分配问题，即确定哪个 Partition 由哪个 Consumer 来消费。Kafka 有两种分配策略，一个是 RoundRobin，一个是 Range，默认为Range。

一个消费者组内也可以订阅多个topic

多个消费组可以订阅同一个topic 。

Zookeeper ：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。

使用brew进行安装，非常方便。

ZooKeeper是一个分布式的，开放源码的 分布式应用程序协调服务 ，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

kafka是基于zookeeper的，启动kafka之前，需要先启动zookeeper

查看启动是否成功

启动kafka

查看启动是否成功

查看topic列表

新起一个终端，作为生产者，用于发送消息，每一行算一条消息，将消息发送到kafka服务器

新起一个终端作为消费者，接收消息

服务关闭的顺序是先kafka，然后zookeeper

再过半小时，你就能明白kafka的工作原理了

Kafka架构原理，也就这么回事！

为什么搭建Kafka需要zookeeper？

需要一个地方存元信息。zookeeper又是分布式，做配置管理比较好的。于是就用上了。

没zookeeper的话，也需要一套机制存储元数据和交换集群信息的工具。

kafka的官方文档有说明。zookeeper是为了解决分布式一致性问题的工具。

至于kafka为什么使用zk，你首先要知道zk的作用, 作为去中心化的集群模式。

需要要消费者知道现在那些生产者（对于消费者而言，kafka就是生产者）是可用的。

如果没了zk消费者如何知道，如果每次消费者在消费之前都去尝试连接生产者测试下是否连接成功，那么效率问题怎么解决。

所以kafka需要zk，在kafka的设计中就依赖了zk了。

Zookeeper 介绍及zookeeper与kafka之间的关系

ZooKeeper

ZooKeeper 是一项集中式服务，用于维护配置信息，命名，提供分布式同步和提供 组 服务。

基本功能：

1、配置管理

　ZooKeeper 为分布式系统提供了一种配置管理的服务：集中管理配置，即将全局配置信息保存在 ZooKeeper 服务中，方便进行修改和管理，省去了手动拷贝配置的过程，同时还保证了可靠和一致性

2、命名服务

分布式系统中，对应用或者服务进行统一命名，用于区分。

3、分布式锁

在分布式系统中，分布式程序分布在各个主机上的进程对互斥资源进行访问时也需要加锁。

分布式锁应当具备以下条件：

在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；

高可用的获取锁与释放锁；

高性能的获取锁与释放锁；

具备可重入特性（可理解为重新进入，由多于一个任务并发使用，而不必担心数据错误）；

具备锁失效机制，防止死锁；

具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败。

4、集群管理

　在分布式系统中，由于各种各样的原因，导致集群中的节点增加或者减少，集群中有些机器需要感知到这种变化，然后根据这种变化做出对应的决策。

ZooKeeper 作为给分布式系统提供协调服务的工具被 kafka 所依赖.

Kafka用Zookeeper所做的那些事

Kafka是高可用的分布式消息系统，首先要解决的就是资源协调分配和多副本状态维护的问题。解决这些问题通常就是两种思路，一是依靠Zookeeper来协调，二是设定一个中心节点，让这个中心节点来协调。如果依靠Zookeeper来协调，会存在大量的竞争条件，对Zookeeper的访问压力增大，而且如果Zookeeper出现了问题（比如网络抖动），系统很容易出现紊乱。Kafka采用的是第二种思路，即选举一个中心节点来进行资源协调与多副本状态维护，这个中心节点被称作Controller（一个特殊的Broker），这个选举过程依靠Zookeeper来完成。

Broker启动时，会竞争创建临时"/controller"。如果创建成功，则成为Controller，并把Broker的id等信息写入这个节点。同时会全程监控"/controller"的数据变化，如果旧的Controller挂掉，则开启新一轮的竞争过程。

Kafka要进行资源协调，第一件需要知道的事情就是各个Broker的存活状态，这个问题利用Zookeeper可以很容易做到。

假设某个Broker，id为0，它启动时，会创建"/brokers/ids/0"临时节点，并把端口等信息写进去。Controller会监控"/brokers/ids"的节点变化，以实时感知各broker的状态，进行资源协调。

在Kafka这个多副本分区的消息系统里，创建一个topic，至少需要以下3个步骤：

Controller的存在，可以很容易完成上面的b和c步骤，但a步骤不行，如果Controller挂掉，则这些信息会不可用。Kafka把这些信息保存在Zookeeper中，依靠其高可用特性来保证这些信息的高可用。假设某个topic名字为mytopic，创建时，其分区信息保存在"/brokers/topics/mytopic"中。Controller全程监控"/brokers/topics"的孩子节点变动，实时感知这些信息，以完成后续步骤。

创建完成之后，后续往往会有分区调整和topic删除等需求。普通青年可能会觉得这两个问题很简单，给Controller发个相关请求就可以了。事实远非如此！

拿分区调整来说，假设某分区有三个副本，分别位于Broker-1、Broker-2和Broker-3，leader为1，现在扩容增加了Broker-4、Broker-5、Broker-6，为了平衡机器间压力，需要将副本1 2 3移到4 5 6，至少经历以下步骤：

以上每个步骤都有可能失败，如何才能保证这次调整顺利进行呢？

首先，我们不能直接修改该分区的副本信息为 4 5 6，原因很简单，需要等待4 5 6的追赶过程以便产生新leader。其次，操作未完全成功的命令需要保存下来，如果操作过程中，Controller挂掉，则新的Controller可以从头开始直至成功。

Kafka怎么做的呢？

当然，这里一次只能有一个调整命令，但一个调整命令可以同时调整多个topic的多个分区。

在这个过程中，Zookeeper的作用是： 持久化操作命令并实时通知操作者 ，是不是只有Zookeeper可以做这个事情呢，不是，但Zookeeper可以做得很好，保证命令高可用。

类似的操作还有topic删除，副本的leader变更等，都是沿用上面的套路。

Broker的集群中有全局配置信息，但如果想针对某个topic或者某个client进行配置呢，Kafka把这些信息保存在Zookeeper中，各个Broker实时监控以更新。

脑裂问题是指，在一个设有中心节点的系统中，出现了两个中心节点。两个中心同时传达命令，自然会造成系统的紊乱。

Kafka利用Zookeeper所做的第一件也是至关重要的一件事情是选举Controller，那么自然就有疑问，有没有可能产生两个Controller呢？

首先，Zookeeper也是有leader的，它有没有可能产生两个leader呢？答案是不会。

quorum机制可以保证，不可能同时存在两个leader获得大多数支持。假设某个leader假死，其余的followers选举出了一个新的leader。这时，旧的leader复活并且仍然认为自己是leader，这个时候它向其他followers发出写请求也是会被拒绝的。因为每当新leader产生时，会生成一个epoch，这个epoch是递增的，followers如果确认了新的leader存在，知道其epoch，就会拒绝epoch小于现任leader epoch的所有请求。那有没有follower不知道新的leader存在呢，有可能，但肯定不是大多数，否则新leader无法产生。Zookeeper的写也遵循quorum机制，因此，得不到大多数支持的写是无效的，旧leader即使各种认为自己是leader，依然没有什么作用。

Kafka的Controller也采用了epoch，具体机制如下:

理论上来说，如果Controller的SessionExpired处理成功，则可以避免双leader，但假设SessionExpire处理意外失效的情况：旧Controller假死，新的Controller创建。旧Controller复活，SessionExpired处理 意外失效 ，仍然认为自己是leader。

这时虽然有两个leader，但没有关系，leader只会发信息给存活的broker（仍然与Zookeeper在Session内的），而这些存活的broker则肯定能感知到新leader的存在，旧leader的请求会被拒绝。

每个Broker有一个metaDataCache，缓存有topic和partition的基本信息，可以正常的生产和消费信息，但不能进行topic的创建、调整和删除等操作。

此外，Broker会不断重试连接。

待续

假设Broker数目为B，topic数目为T，所有topic总partition数目为P，Client数目为C，以下数值均为峰值：

Zookeeper 在 Kafka 中的作用

如上图所示，kafaka集群的 broker，和 Consumer 都需要连接 Zookeeper。

Producer 直接连接 Broker。

Producer 把数据上传到 Broker，Producer可以指定数据有几个分区、几个备份。上面的图中，数据有两个分区 0、1，每个分区都有自己的副本：0'、 1'。

黄色的分区为 leader，白色的为 follower。

leader 处理 partition 的所有读写请求，与此同时，follower会被动定期地去复制leader上的数据。 如下图所示，红色的为 leader，绿色的为 follower，leader复制自己到其他 Broker 中：

Topic 分区被放在不同的 Broker 中，保证 Producer 和 Consumer 错开访问 Broker，避免访问单个 Broker造成过度的IO压力，使得负载均衡。

Broker是分布式部署并且相互之间相互独立，但是需要有一个注册系统能够将整个集群中的Broker管理起来 ，此时就使用到了Zookeeper。在Zookeeper上会有一个专门 用来进行Broker服务器列表记录 的节点：

/brokers/ids

每个Broker在启动时，都会到Zookeeper上进行注册，即到/brokers/ids下创建属于自己的节点，如/brokers/ids/[0...N]。

Kafka使用了全局唯一的数字来指代每个Broker服务器，不同的Broker必须使用不同的Broker ID进行注册，创建完节点后， 每个Broker就会将自己的IP地址和端口信息记录 到该节点中去。其中，Broker创建的节点类型是临时节点，一旦Broker宕机，则对应的临时节点也会被自动删除。

在Kafka中，同一个 Topic的消息会被分成多个分区 并将其分布在多个Broker上， 这些分区信息及与Broker的对应关系 也都是由Zookeeper在维护，由专门的节点来记录，如：

/borkers/topics

Kafka中每个Topic都会以/brokers/topics/[topic]的形式被记录，如/brokers/topics/login和/brokers/topics/search等。Broker服务器启动后，会到对应Topic节点（/brokers/topics）上注册自己的Broker ID并写入针对该Topic的分区总数，如/brokers/topics/login/3-2，这个节点表示Broker ID为3的一个Broker服务器，对于"login"这个Topic的消息，提供了2个分区进行消息存储，同样，这个分区节点也是临时节点。

由于同一个Topic消息会被分区并将其分布在多个Broker上，因此， 生产者需要将消息合理地发送到这些分布式的Broker上 ，那么如何实现生产者的负载均衡，Kafka支持传统的四层负载均衡，也支持Zookeeper方式实现负载均衡。

(1) 四层负载均衡，根据生产者的IP地址和端口来为其确定一个相关联的Broker。通常，一个生产者只会对应单个Broker，然后该生产者产生的消息都发往该Broker。这种方式逻辑简单，每个生产者不需要同其他系统建立额外的TCP连接，只需要和Broker维护单个TCP连接即可。但是，其无法做到真正的负载均衡，因为实际系统中的每个生产者产生的消息量及每个Broker的消息存储量都是不一样的，如果有些生产者产生的消息远多于其他生产者的话，那么会导致不同的Broker接收到的消息总数差异巨大，同时，生产者也无法实时感知到Broker的新增和删除。

(2) 使用Zookeeper进行负载均衡，由于每个Broker启动时，都会完成Broker注册过程，生产者会通过该节点的变化来动态地感知到Broker服务器列表的变更，这样就可以实现动态的负载均衡机制。

与生产者类似，Kafka中的消费者同样需要进行负载均衡来实现多个消费者合理地从对应的Broker服务器上接收消息，每个消费者分组包含若干消费者， 每条消息都只会发送给分组中的一个消费者 ，不同的消费者分组消费自己特定的Topic下面的消息，互不干扰。

消费组 (Consumer Group)：

consumer group 下有多个 Consumer（消费者）。

对于每个消费者组 (Consumer Group)，Kafka都会为其分配一个全局唯一的Group ID，Group 内部的所有消费者共享该 ID。订阅的topic下的每个分区只能分配给某个 group 下的一个consumer(当然该分区还可以被分配给其他group)。

同时，Kafka为每个消费者分配一个Consumer ID，通常采用"Hostname:UUID"形式表示。

在Kafka中，规定了 每个消息分区 只能被同组的一个消费者进行消费 ，因此，需要在 Zookeeper 上记录 消息分区 与 Consumer 之间的关系，每个消费者一旦确定了对一个消息分区的消费权力，需要将其Consumer ID 写入到 Zookeeper 对应消息分区的临时节点上，例如：

/consumers/[group_id]/owners/[topic]/[broker_id-partition_id]

其中，[broker_id-partition_id]就是一个 消息分区 的标识，节点内容就是该 消息分区 上 消费者的Consumer ID。

在消费者对指定消息分区进行消息消费的过程中， 需要定时地将分区消息的消费进度Offset记录到Zookeeper上 ，以便在该消费者进行重启或者其他消费者重新接管该消息分区的消息消费后，能够从之前的进度开始继续进行消息消费。Offset在Zookeeper中由一个专门节点进行记录，其节点路径为:

/consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]

节点内容就是Offset的值。

消费者服务器在初始化启动时加入消费者分组的步骤如下

注册到消费者分组。每个消费者服务器启动时，都会到Zookeeper的指定节点下创建一个属于自己的消费者节点，例如/consumers/[group_id]/ids/[consumer_id]，完成节点创建后，消费者就会将自己订阅的Topic信息写入该临时节点。

对 消费者分组 中的 消费者 的变化注册监听 。每个 消费者 都需要关注所属 消费者分组 中其他消费者服务器的变化情况，即对/consumers/[group_id]/ids节点注册子节点变化的Watcher监听，一旦发现消费者新增或减少，就触发消费者的负载均衡。

对Broker服务器变化注册监听 。消费者需要对/broker/ids/[0-N]中的节点进行监听，如果发现Broker服务器列表发生变化，那么就根据具体情况来决定是否需要进行消费者负载均衡。

进行消费者负载均衡 。为了让同一个Topic下不同分区的消息尽量均衡地被多个 消费者 消费而进行 消费者 与 消息 分区分配的过程，通常，对于一个消费者分组，如果组内的消费者服务器发生变更或Broker服务器发生变更，会发出消费者负载均衡。

以下是kafka在zookeep中的详细存储结构图：

早期版本的 kafka 用 zk 做 meta 信息存储，consumer 的消费状态，group 的管理以及 offse t的值。考虑到zk本身的一些因素以及整个架构较大概率存在单点问题，新版本中确实逐渐弱化了zookeeper的作用。新的consumer使用了kafka内部的group coordination协议，也减少了对zookeeper的依赖