实时计算之kafka的高可用保证

1、kakfa集群介绍01kakfa服务的故障与恢复02kafka的HA机制03CONTENTSkafka集群简单介绍kafka集群有4个kakfa集群cdn集群：40台user-action集群：30台性能测试集群：10台cdn老集群：60台 （即将淘汰）kafka服务的不稳定因素v 磁盘损坏会造成kafka服务异常退出。v 服务器故障kafka服务退出；两台以上故障：kafka数据丢失等问题。v zookeeper服务之前所有的kafka和共用一套zookeeper服务，现已经分拆。kafka服务器进程监控目前的监控方式：使用crontab定时扫描监测kafka服务进程，如kafka服务不存在

2、，自动拉起，并短信报警给相关人员。缺点：时效性不高。其他方式：daemontools监控。KAFKA磁盘故障自动恢复由于kafka服务器磁盘开销很大，随着集群规模的增大，出现磁盘故障的可能性大大提高，为了减少维护的成本，我们开发了磁盘故障的自动处理程序。v通过mysql来管理损坏磁盘；v自动剔除故障磁盘。v自动恢复磁盘v服务器运行日志目录重新选举。v磁盘故障报警通知。kafka备份池kafka备份池：安装了3台kafka服务，平时不启动，线上kafka服务器故障时，从备份池中取出一台备份服务器接替，同时通知IT及时维修故障的服务器。由于目前的kafka的topic副本数为2；当kafka服务器

3、故障超过两台时，就会造成数据丢失；当kafka服务器故障时，可根据需要启动备份服务器。kafka备份机策略可行性测试检验备份机策略是否可行，同时为了检验服务器故障对服务的影响，检验kafka备份机切换对集群和服务的影响，做了如下测试。测试方案： step1:部署3台测试集群的kafka集群备份机。 step2: 创建一个测试的topic，只有一个分区，备份数为2 的test_bak_kafka step3: 编写一个发送消息的生产者程序，不停的发送消息， 启动一个消费者，实时消费消息。 step4:关闭一个正在写入消息的broker的主节点 stop5:启动被关闭的节点。 step6: 关闭主

4、节点，启动 备份节点 kafka的zk元数据信息介绍关闭topic分区的leader节点 现象：现象：生产者写消息的生产者会报如下异常退去，不可恢复。上层应用需要考虑重试机制消费者：不受影响。kafka服务器：topic的备份节点切换成了leader当异常的broker节点退出时,leader也会重新选举，而异常节点加入时，不会重新选举。如果故障退出的为follower节点，则生产者和消费者都不会受到影响。启动备份节点步骤步骤操作操作结果结果stop1：修改备份节点的borker.id为故障节点ID需提前配置备份节点到/etc/hoststop2： 启动备份节点正常启动stop3：观察zk中b

5、rokerid的变化zk中/brokers/ids 的出现brokerid的节点stop4：观察生产者和消费者的状态没有影响stop5：观察备份节点的数据同步状态数据没有同步过来，等待一段时间后，可以恢复同步。数据同步效率：1个磁盘：16分钟 53G的数据。 54M/stopic的zk的状态变化在kafka节点异常，备份节点恢复过程中，topic的zk状态变化如下：关闭主节点：kafka的leader发生切换启动备份节点：正常状态：一台broker故障后：备份broker恢复后：kafka的Leader Electionkafka的partition的leader算法没有使用常用的“Major

6、ity Vote”；而是采用在ZK中动态维护了一个ISR，这个ISR里的所有Replica都需要跟上了leader，只有ISR里的成员才有被选为Leader的可能。优点：优点：vN+1个Replica，容忍N个replica失败v容忍N个Replica的失败，比“Majority Vote”少近一半的replica缺点：缺点：需要等待最慢的Broker，但是可以通过Producer选择是否被commit阻塞来改善（request.required.acks）。kafka的Leader Election实现：kafka通过Controller节点来选举各个partition的leader。kaf

7、ka的Leader Election1.controller在ZK的/brokers/ids节点上注册Watch2.broker宕机，ZK会fire Controller注册的Watch会触发controller。3.Controller重新设置partition信息（set_p）。4.从/brokers/topics/topic/partitions/partition/state读取该Partition当前的ISR。5.决定该Partition的新Leader如果ISR 列表中有Replica有幸存，选择其中一个为新Leader，新ISR为当前ISR；如果ISR列表中的Replica都当宕

8、掉，选择任意一个幸存的Replica为新Leader，以及ISR（会有数据丢失的可能）；如果该Partition的所有Replica都宕机了，则将新的Leader设置为-1；6. Controlle直接通过RPC向相关Broker发送LeaderAndISRRequest命令。如何处理所有Replica都不工作如果某个Partition的所有Replica都宕机了，就无法保证数据不丢失了。这种情况下有两种可行的方案：1.等待ISR中的任一个Replica“活”过来，并且选它作为Leader 2.选择第一个“活”过来的Replica（不一定是ISR中的）作为Leader。可用性和一致性当中一个简

9、单的折衷，kafka使用的第二种方法。kafka的Data Replication消息生产者的消息发送机制：同步发送同步发送: Producer会在尝试重新发送message.send.max.retries 次后抛出Exception异步发送：异步发送： Producer会尝试重新发送message.send.max.retries 次后记录该异常并继续发送后续数据。目的：kafka为预防一旦某一个Broker宕机，则其上所有的Partition数据都不可被消费；同时提高了整个系统的可用性。vReplica均匀分布到整个集群 Kafka分配Replica的算法如下：设有n个broker，i个

10、partion，j个Replica，将broker和parttion排序后。 将第i个Partition分配到第（i mod n）个Broker上 将第i个Partition的第j个Replica分配到第（(i + j) mod n）个Broker上kafka的数据同步机制kafka的各个分区备份之间的同步机制：1.生产者发送消息到leader，leader log数据2. flowller从leader拉取消息。3.flower接收到消息，放到内存ack leader。4.leader ack producer。Kafka之Replication toolsu最优副本选举工具bin/kafk

11、a-preferred-replica-election.s -zookeeper localhost:12913/kafka -path-to-json-file topicPartitionList.jsonuTopic 和Partion转移工具#转移topic到新的brokerkafka-reassign-partitions.sh -topics-to-move-json-file topics-to-move.json -broker-list 5,6,7 -execute #转移partition的一些replica到指定的brokerkafka-reassign-partitions.sh -manual-assignment-json-file partitions-to-move.json -execute kafka磁盘上限测试kafka磁盘容量测试，当一个topic的一个分区所在的磁盘用尽，是否会异常？测试环境：测试环境：1.开启一个写kafka的生生产者，向kafka写消息，每条消息10kb。2.Topic为1个partitions，2个Replica；Replica对应的磁盘大小为5.5结果