模块化网站建设 局域网,做竞价的网站有利于优化吗,池州网站建设开发,软件工程师要求正文 一、高可用的由来 1.1 为何需要Replication 在Kafka在0.8以前的版本中#xff0c;是没有Replication的#xff0c;一旦某一个Broker宕机#xff0c;则其上所有的Partition数据都不可被消费#xff0c;这与Kafka数据持久性及Delivery Guarantee的设计目标相悖。同时Pr…正文 一、高可用的由来 1.1 为何需要Replication 在Kafka在0.8以前的版本中是没有Replication的一旦某一个Broker宕机则其上所有的Partition数据都不可被消费这与Kafka数据持久性及Delivery Guarantee的设计目标相悖。同时Producer都不能再将数据存于这些Partition中。 如果Producer使用同步模式则Producer会在尝试重新发送message.send.max.retries默认值为3次后抛出Exception用户可以选择停止发送后续数据也可选择继续选择发送。而前者会造成数据的阻塞后者会造成本应发往该Broker的数据的丢失。 如果Producer使用异步模式则Producer会尝试重新发送message.send.max.retries默认值为3次后记录该异常并继续发送后续数据这会造成数据丢失并且用户只能通过日志发现该问题。同时Kafka的Producer并未对异步模式提供callback接口。 由此可见在没有Replication的情况下一旦某机器宕机或者某个Broker停止工作则会造成整个系统的可用性降低。随着集群规模的增加整个集群中出现该类异常的几率大大增加因此对于生产系统而言Replication机制的引入非常重要。 1.2 Leader Election 引入Replication之后同一个Partition可能会有多个Replica而这时需要在这些Replication之间选出一个LeaderProducer和Consumer只与这个Leader交互其它Replica作为Follower从Leader中复制数据。 因为需要保证同一个Partition的多个Replica之间的数据一致性其中一个宕机后其它Replica必须要能继续服务并且即不能造成数据重复也不能造成数据丢失。如果没有一个Leader所有Replica都可同时读/写数据那就需要保证多个Replica之间互相N×N条通路同步数据数据的一致性和有序性非常难保证大大增加了Replication实现的复杂性同时也增加了出现异常的几率。而引入Leader后只有Leader负责数据读写Follower只向Leader顺序Fetch数据N条通路系统更加简单且高效。 二、Kafka HA设计解析 2.1 如何将所有Replica均匀分布到整个集群 为了更好的做负载均衡Kafka尽量将所有的Partition均匀分配到整个集群上。一个典型的部署方式是一个Topic的Partition数量大于Broker的数量。同时为了提高Kafka的容错能力也需要将同一个Partition的Replica尽量分散到不同的机器。实际上如果所有的Replica都在同一个Broker上那一旦该Broker宕机该Partition的所有Replica都无法工作也就达不到HA的效果。同时如果某个Broker宕机了需要保证它上面的负载可以被均匀的分配到其它幸存的所有Broker上。
Kafka分配Replica的算法如下
1.将所有Broker假设共n个Broker和待分配的Partition排序
2.将第i个Partition分配到第i mod n个Broker上
3.将第i个Partition的第j个Replica分配到第(i j) mode n个Broker上 2.2 Data Replication副本策略 Kafka的高可靠性的保障来源于其健壮的副本replication策略。 2.2.1 消息传递同步策略 Producer在发布消息到某个Partition时先通过ZooKeeper找到该Partition的Leader然后无论该Topic的Replication Factor为多少Producer只将该消息发送到该Partition的Leader。Leader会将该消息写入其本地Log。每个Follower都从Leader pull数据。这种方式上Follower存储的数据顺序与Leader保持一致。Follower在收到该消息并写入其Log后向Leader发送ACK。一旦Leader收到了ISR中的所有Replica的ACK该消息就被认为已经commit了Leader将增加HW并且向Producer发送ACK。
为了提高性能每个Follower在接收到数据后就立马向Leader发送ACK而非等到数据写入Log中。因此对于已经commit的消息Kafka只能保证它被存于多个Replica的内存中而不能保证它们被持久化到磁盘中也就不能完全保证异常发生后该条消息一定能被Consumer消费。
Consumer读消息也是从Leader读取只有被commit过的消息才会暴露给Consumer。
Kafka Replication的数据流如下图所示 2.2.2 ACK前需要保证有多少个备份 对于Kafka而言定义一个Broker是否“活着”包含两个条件
一是它必须维护与ZooKeeper的session这个通过ZooKeeper的Heartbeat机制来实现。二是Follower必须能够及时将Leader的消息复制过来不能“落后太多”。
Leader会跟踪与其保持同步的Replica列表该列表称为ISR即in-sync Replica。如果一个Follower宕机或者落后太多Leader将把它从ISR中移除。这里所描述的“落后太多”指Follower复制的消息落后于Leader后的条数超过预定值该值可在$KAFKA_HOME/config/server.properties中通过replica.lag.max.messages配置其默认值是4000或者Follower超过一定时间该值可在$KAFKA_HOME/config/server.properties中通过replica.lag.time.max.ms来配置其默认值是10000未向Leader发送fetch请求。
Kafka的复制机制既不是完全的同步复制也不是单纯的异步复制。事实上完全同步复制要求所有能工作的Follower都复制完这条消息才会被认为commit这种复制方式极大的影响了吞吐率高吞吐率是Kafka非常重要的一个特性。而异步复制方式下Follower异步的从Leader复制数据数据只要被Leader写入log就被认为已经commit这种情况下如果Follower都复制完都落后于Leader而如果Leader突然宕机则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。Follower可以批量的从Leader复制数据这样极大的提高复制性能批量写磁盘极大减少了Follower与Leader的差距。
需要说明的是Kafka只解决fail/recover不处理“Byzantine”“拜占庭”问题。一条消息只有被ISR里的所有Follower都从Leader复制过去才会被认为已提交。这样就避免了部分数据被写进了Leader还没来得及被任何Follower复制就宕机了而造成数据丢失Consumer无法消费这些数据。而对于Producer而言它可以选择是否等待消息commit这可以通过request.required.acks来设置。这种机制确保了只要ISR有一个或以上的Follower一条被commit的消息就不会丢失。 2.2.3 Leader Election算法 Leader选举本质上是一个分布式锁有两种方式实现基于ZooKeeper的分布式锁
节点名称唯一性多个客户端创建一个节点只有成功创建节点的客户端才能获得锁临时顺序节点所有客户端在某个目录下创建自己的临时顺序节点只有序号最小的才获得锁
一种非常常用的选举leader的方式是“Majority Vote”“少数服从多数”但Kafka并未采用这种方式。这种模式下如果我们有2f1个Replica包含Leader和Follower那在commit之前必须保证有f1个Replica复制完消息为了保证正确选出新的Leaderfail的Replica不能超过f个。因为在剩下的任意f1个Replica里至少有一个Replica包含有最新的所有消息。这种方式有个很大的优势系统的latency只取决于最快的几个Broker而非最慢那个。Majority Vote也有一些劣势为了保证Leader Election的正常进行它所能容忍的fail的follower个数比较少。如果要容忍1个follower挂掉必须要有3个以上的Replica如果要容忍2个Follower挂掉必须要有5个以上的Replica。也就是说在生产环境下为了保证较高的容错程度必须要有大量的Replica而大量的Replica又会在大数据量下导致性能的急剧下降。这就是这种算法更多用在ZooKeeper这种共享集群配置的系统中而很少在需要存储大量数据的系统中使用的原因。例如HDFS的HA Feature是基于majority-vote-based journal但是它的数据存储并没有使用这种方式。
Kafka在ZooKeeper中动态维护了一个ISRin-sync replicas这个ISR里的所有Replica都跟上了leader只有ISR里的成员才有被选为Leader的可能。在这种模式下对于f1个Replica一个Partition能在保证不丢失已经commit的消息的前提下容忍f个Replica的失败。在大多数使用场景中这种模式是非常有利的。事实上为了容忍f个Replica的失败Majority Vote和ISR在commit前需要等待的Replica数量是一样的但是ISR需要的总的Replica的个数几乎是Majority Vote的一半。
虽然Majority Vote与ISR相比有不需等待最慢的Broker这一优势但是Kafka作者认为Kafka可以通过Producer选择是否被commit阻塞来改善这一问题并且节省下来的Replica和磁盘使得ISR模式仍然值得。 2.2.4 如何处理所有Replica都不工作 在ISR中至少有一个follower时Kafka可以确保已经commit的数据不丢失但如果某个Partition的所有Replica都宕机了就无法保证数据不丢失了。这种情况下有两种可行的方案
1.等待ISR中的任一个Replica“活”过来并且选它作为Leader
2.选择第一个“活”过来的Replica不一定是ISR中的作为Leader
这就需要在可用性和一致性当中作出一个简单的折衷。如果一定要等待ISR中的Replica“活”过来那不可用的时间就可能会相对较长。而且如果ISR中的所有Replica都无法“活”过来了或者数据都丢失了这个Partition将永远不可用。选择第一个“活”过来的Replica作为Leader而这个Replica不是ISR中的Replica那即使它并不保证已经包含了所有已commit的消息它也会成为Leader而作为consumer的数据源前文有说明所有读写都由Leader完成。Kafka0.8.*使用了第二种方式。根据Kafka的文档在以后的版本中Kafka支持用户通过配置选择这两种方式中的一种从而根据不同的使用场景选择高可用性还是强一致性。 2.2.5 选举Leader 最简单最直观的方案是所有Follower都在ZooKeeper上设置一个Watch一旦Leader宕机其对应的ephemeral znode会自动删除此时所有Follower都尝试创建该节点而创建成功者ZooKeeper保证只有一个能创建成功即是新的Leader其它Replica即为Follower。
但是该方法会有3个问题
1.split-brain 这是由ZooKeeper的特性引起的虽然ZooKeeper能保证所有Watch按顺序触发但并不能保证同一时刻所有Replica“看”到的状态是一样的这就可能造成不同Replica的响应不一致
2.herd effect 如果宕机的那个Broker上的Partition比较多会造成多个Watch被触发造成集群内大量的调整
3.ZooKeeper负载过重 每个Replica都要为此在ZooKeeper上注册一个Watch当集群规模增加到几千个Partition时ZooKeeper负载会过重。
Kafka 0.8.*的Leader Election方案解决了上述问题它在所有broker中选出一个controller所有Partition的Leader选举都由controller决定。controller会将Leader的改变直接通过RPC的方式比ZooKeeper Queue的方式更高效通知需为为此作为响应的Broker。同时controller也负责增删Topic以及Replica的重新分配。 三、HA相关ZooKeeper结构 3.1 admin 该目录下znode只有在有相关操作时才会存在操作结束时会将其删除
/admin/reassign_partitions用于将一些Partition分配到不同的broker集合上。对于每个待重新分配的PartitionKafka会在该znode上存储其所有的Replica和相应的Broker id。该znode由管理进程创建并且一旦重新分配成功它将会被自动移除。 3.2 broker 即/brokers/ids/[brokerId]存储“活着”的broker信息。
topic注册信息/brokers/topics/[topic]存储该topic的所有partition的所有replica所在的broker id第一个replica即为preferred replica对一个给定的partition它在同一个broker上最多只有一个replica,因此broker id可作为replica id。 3.3 controller /controller - int (broker id of the controller)存储当前controller的信息
/controller_epoch - int (epoch)直接以整数形式存储controller epoch而非像其它znode一样以JSON字符串形式存储。 四、producer发布消息 4.1 写入方式 producer 采用 push 模式将消息发布到 broker每条消息都被 append 到 patition 中属于顺序写磁盘顺序写磁盘效率比随机写内存要高保障 kafka 吞吐率。 4.2 消息路由 producer 发送消息到 broker 时会根据分区算法选择将其存储到哪一个 partition。其路由机制为
1、 指定了 patition则直接使用
2、 未指定 patition 但指定 key通过对 key 的 value 进行hash 选出一个 patition
3、 patition 和 key 都未指定使用轮询选出一个 patition。 4.3 写入流程 producer 写入消息序列图如下所示 流程说明 1、 producer 先从 zookeeper 的 /brokers/.../state 节点找到该 partition 的 leader
2、 producer 将消息发送给该 leader
3、 leader 将消息写入本地 log
4、 followers 从 leader pull 消息写入本地 log 后 leader 发送 ACK
5、 leader 收到所有 ISR 中的 replica 的 ACK 后增加 HWhigh watermark最后 commit 的 offset 并向 producer 发送 ACK 五、broker保存消息 5.1 存储方式 物理上把 topic 分成一个或多个 patition对应 server.properties 中的 num.partitions3 配置每个 patition 物理上对应一个文件夹该文件夹存储该 patition 的所有消息和索引文件如下 5.2 存储策略 无论消息是否被消费kafka 都会保留所有消息。有两种策略可以删除旧数据
1、 基于时间log.retention.hours168
2、 基于大小log.retention.bytes1073741824 六、Topic的创建和删除 6.1 创建topic 创建 topic 的序列图如下所示 流程说明 1、 controller 在 ZooKeeper 的 /brokers/topics 节点上注册 watcher当 topic 被创建则 controller 会通过 watch 得到该 topic 的 partition/replica 分配。
2、 controller从 /brokers/ids 读取当前所有可用的 broker 列表对于 set_p 中的每一个 partition2.1、 从分配给该 partition 的所有 replica称为AR中任选一个可用的 broker 作为新的 leader并将AR设置为新的 ISR 2.2、 将新的 leader 和 ISR 写入 /brokers/topics/[topic]/partitions/[partition]/state
3、 controller 通过 RPC 向相关的 broker 发送 LeaderAndISRRequest。 6.2 删除topic 删除 topic 的序列图如下所示 流程说明 1、 controller 在 zooKeeper 的 /brokers/topics 节点上注册 watcher当 topic 被删除则 controller 会通过 watch 得到该 topic 的 partition/replica 分配。
2、 若 delete.topic.enablefalse结束否则 controller 注册在 /admin/delete_topics 上的 watch 被 firecontroller 通过回调向对应的 broker 发送 StopReplicaRequest。 七、broker failover
kafka broker failover 序列图如下所示 流程说明 1、 controller 在 zookeeper 的 /brokers/ids/[brokerId] 节点注册 Watcher当 broker 宕机时 zookeeper 会 fire watch
2、 controller 从 /brokers/ids 节点读取可用broker
3、 controller决定set_p该集合包含宕机 broker 上的所有 partition
4、 对 set_p 中的每一个 partition 4.1、 从/brokers/topics/[topic]/partitions/[partition]/state 节点读取 ISR 4.2、 决定新 leader 4.3、 将新 leader、ISR、controller_epoch 和 leader_epoch 等信息写入 state 节点
5、 通过 RPC 向相关 broker 发送 leaderAndISRRequest 命令 八、controller failover
当 controller 宕机时会触发 controller failover。每个 broker 都会在 zookeeper 的 /controller 节点注册 watcher当 controller 宕机时 zookeeper 中的临时节点消失所有存活的 broker 收到 fire 的通知每个 broker 都尝试创建新的 controller path只有一个竞选成功并当选为 controller。
当新的 controller 当选时会触发 KafkaController.onControllerFailover 方法在该方法中完成如下操作 1、 读取并增加 Controller Epoch。
2、 在 reassignedPartitions Patch(/admin/reassign_partitions) 上注册 watcher。
3、 在 preferredReplicaElection Path(/admin/preferred_replica_election) 上注册 watcher。
4、 通过 partitionStateMachine 在 broker Topics Patch(/brokers/topics) 上注册 watcher。
5、 若 delete.topic.enabletrue默认值是 false则 partitionStateMachine 在 Delete Topic Patch(/admin/delete_topics) 上注册 watcher。
6、 通过 replicaStateMachine在 Broker Ids Patch(/brokers/ids)上注册Watch。
7、 初始化 ControllerContext 对象设置当前所有 topic“活”着的 broker 列表所有 partition 的 leader 及 ISR等。
8、 启动 replicaStateMachine 和 partitionStateMachine。
9、 将 brokerState 状态设置为 RunningAsController。
10、 将每个 partition 的 Leadership 信息发送给所有“活”着的 broker。
11、 若 auto.leader.rebalance.enabletrue默认值是true则启动 partition-rebalance 线程。
12、 若 delete.topic.enabletrue 且Delete Topic Patch(/admin/delete_topics)中有值则删除相应的Topic。
