Kafka 调研

2019-03-24 本文已影响0人王龙江_3c83

1. 基本概念

术语	解释
Producer
Consumer
Consumer Group
Broker
Controller
Topic	相同Topic的生产者和消费者进行通信。
offset	某个user group在某个partiton中当前已经消费到达的位置。
Partition	分区
Leader	负责读写指定分区的节点
Replicas	复制该分区log的节点列表
Isr	"in-sync" replicas，当前活跃的副本列表（是一个子集），并且可能成为Leader

2. 重要配置

2.1 有序性保证

对于有序性要求严格的场景，将 retries 时间设置为 Broker 主从切换时间，次数设置为合适的正数，将 max.in.flight.requests.per.connection设置为 1。

2.2 可靠性保证

尽量避免单个硬件（电源连接、交换机）故障导致多台 broker 停止服务。
假设每次故障仅会发生在单台 broker 的前提下，进行如下设置：（1）将 request.required.acks 设置为 -1，等待所有 producer 等待 ISR 中的所有 replica 的 ack。（2）isr replica 的确认，慎重设置，设置过于严格导致误判。（3）min.insync.replicas：ISR中最少的follower个数。一般设置成大于等于 2 的，否则如果最后的一个broker挂了，则数据就丢失了。

2.3 分区再均衡

数据之间不存在逻辑上依赖性，写数据库后写 zk，并且采用数据库的 UUID 保证只写一次。

1. zookeeper 在 kafka 中起到什么作用

1.1 Controller 选举

Controller 是一个特殊的 Broker, 其负责维护所有 Partition 的 leader/follower 关系。当有 partition 的 leader 挂掉之后，controller 会重新从同步队列中选出一个 leader。
Zookeeper 负责从 Broker 中选举出一个作为 Controller, 并确保其唯一性。同时, 当Controller 宕机时, 选举一个新的。

1.2 集群 membership

记录集群中都有哪些活跃着的Broker。

1.3 Topic 配置

记录有哪些Topic, Topic 都有哪些 Partition，Replica 存放在哪里, Leader 是谁。

1.4 在 consumer group 发生变化时进行 rebalance。

1.5 配额(0.9.0+)

记录每个客户能够读写的数据量。

1.6 ACLs(0.9.0+)

记录对Topic 的读写控制。

1.7 high-level consumer(已废弃)

记录consumer group 及其成员和offset 信息。

2. kafka 在 zookeeper 上创建的目录结构

image

详细内容参考链接：
kafka笔记-Kafka在zookeeper中的存储结构【转】

注册的节点如下：

consumers、admin、config、controller、brokers、controller_epoch

topic 注册信息

/brokers/topics/[topic]：存储某个 topic 的 partitions 所有分配信息
partition状态信息

/brokers/topics/[topic]/partitions/[0...N] 其中[0..N]表示partition索引号

/brokers/topics/[topic]/partitions/[partitionId]/state
Broker注册信息

/brokers/ids/[0...N]

每个broker的配置文件中都需要指定一个数字类型的id(全局不可重复),此节点为临时znode(EPHEMERAL)
Controller epoch

/controller_epoch -> int (epoch)

此值为一个数字,kafka集群中第一个broker第一次启动时为1，以后只要集群中center controller中央控制器所在broker变更或挂掉，就会重新选举新的center controller，每次center controller变更controller_epoch值就会 + 1
Controller注册信息

/controller -> int (broker id of the controller)
存储center controller中央控制器所在kafka broker的信息。这个值默认是 1，当 controller 节点挂掉后重新选举 controller 后，值会 +1
Consumer注册信息
/consumers/[groupId]/ids/[consumerIdString]
每个consumer都有一个唯一的ID(consumerId可以通过配置文件指定,也可以由系统生成),此id用来标记消费者信息
Consumer owner

/consumers/[groupId]/owners/[topic]/[partitionId] -> consumerIdString + threadId索引编号
记录当前组下每个 topic 的 partionId 被哪个消费者的线程消费。

3. kafka 中的控制器 controller 的作用

Kafka集群中的其中一个 Broker 会被选举为Controller，主要负责 Partition 管理和副本状态管理（当 partition leader 挂掉时，会从 follower 中选出一个 leader），也会执行类似于重分配 Partition 之类的管理任务。如果当前的 Controller 失败，会从其他正常的 Broker 中重新选举 Controller。

4. kafka consumer 均衡算法

当一个group中,有consumer加入或者离开时，会触发partitions均衡。均衡的最终目的，是提升topic的并发消费能力。

假如 topic1 具有如下 partitions: P0,P1,P2,P3
假如 group 中有如下 consumer: C0,C1
首先根据 partition 索引号对 partitions 排序: P0,P1,P2,P3
根据(consumer.id + '-'+ thread序号)排序: C0,C1
计算倍数: M = [P0,P1,P2,P3].size / [C0,C1].size，本例值 M = 2 (向上取整)
然后依次分配 partitions: C0 = [P0,P1],C1=[P2,P3]，即 Ci = [P(i * M),P((i + 1) * M -1)]

5. kafka 数据高可用的原理是什么

一致性定义:若某条消息对Consumer可见,那么即使Leader宕机了,在新Leader上数据依然可以被读到

HighWaterMark简称HW: Partition的高水位，取一个partition对应的ISR中最小的LEO作为HW，消费者最多只能消费到HW所在的位置，另外每个replica都有highWatermark，leader和follower各自负责更新自己的highWatermark状态，highWatermark <= leader. LogEndOffset
对于Leader新写入的msg，Consumer不能立刻消费，Leader会等待该消息被所有ISR中的replica同步后,更新HW,此时该消息才能被Consumer消费，即Consumer最多只能消费到HW位置

这样就保证了如果Leader Broker失效,该消息仍然可以从新选举的Leader中获取。对于来自内部Broker的读取请求,没有HW的限制。同时,Follower也会维护一份自己的HW,Folloer.HW = min(Leader.HW, Follower.offset)

6. kafka 的数据可靠性保证

当Producer向Leader发送数据时,可以通过acks参数设置数据可靠性的级别

0: 不论写入是否成功,server不需要给Producer发送Response,如果发生异常,server会终止连接,触发Producer更新meta数据;
1: Leader写入成功后即发送Response,此种情况如果Leader fail,会丢失数据
-1: 等待所有ISR接收到消息后再给Producer发送Response,这是最强保证

仅设置acks=-1也不能保证数据不丢失,当Isr列表中只有Leader时,同样有可能造成数据丢失。要保证数据不丢除了设置acks=-1, 还要保证ISR的大小大于等于2,具体参数设置:

1.request.required.acks:设置为-1 等待所有ISR列表中的Replica接收到消息后采算写成功;

2.min.insync.replicas: 设置为大于等于2,保证ISR中至少有两个Replica

注意：Producer要在吞吐率和数据可靠性之间做一个权衡

7. kafka partition 分区的策略是什么

消息发送到哪个分区上，有两种基本的策略，一是采用 Key Hash 算法，一是采用 Round Robin 算法。另外创建分区时，最好是 broker 数量的整数倍，这样才能是一个 Topic 的分区均匀的分布在整个 Kafka 集群中。

默认情况下，Kafka 根据传递消息的 key 来进行分区的分配，即 hash(key) % numPartitions。

如果发送消息时没有指定key，那么 Producer 将会把这条消息发送给随机的一个 Partition。但是代码层面的逻辑并不完全是这样。首先看看Kafka有没有缓存的现成的分区Id，如果有的话直接使用这个分区Id。如果没有的话，找出所有可用分区的leader所在的broker，从中随机挑一个并放到缓存中，下次就直接从缓存中拿这个 partition id。注意这个缓存是每隔一段时间就会被清空的。这么做的目的是为了减少服务器端的sockets数。

8. Kafka Producer是如何动态感知Topic分区数变化

过一定的时间(topic.metadata.refresh.interval.ms参数决定)才知道分区数改变的。