kafka详细原理讲解（三）

一、如何做到高可用：
　　1）创建副本
　　　　如果没有多副本的，一旦某一个Broker宕机，则其上所有的Partition数据都不可被消费，同时Producer都不能再将数据存于这些Partition中。
　　　　如果Producer使用同步模式则Producer会在尝试重新发送message.send.max.retries（默认值为3）次后抛出Exception，用户可以选择停止发送后续数据也可选择继续选择发送。而前者会造成数据的阻塞，后者会造成本应发往该Broker的数据的丢失。
　　　　如果Producer使用异步模式，则Producer会尝试重新发送一定次数后记录该异常并继续发送后续数据，这会造成数据丢失并且用户只能通过日志发现该问题。
　　　　由此可见，在没有多副本的情况下，一旦某机器宕机或者某个Broker停止工作则会造成整个系统的可用性降低。随着集群规模的增加，整个集群中出现该类异常的几率大大增加，因此对于生产系统而言Replication机制的引入非常重要。
　　2）Leader选择
　　　　引入多副本之后，同一个Partition可能会有多个副本，而这时需要在这些多副本之间选出一个Leader，Producer和Consumer只与这个Leader交互，其它副本作为Follower从Leader中复制数据。
　　　　因为需要保证同一个Partition的多个副本之间的数据一致性（其中一个宕机后其它副本必须要能继续服务并且即不能造成数据重复也不能造成数据丢失）。如果没有一个Leader，所有副本都可同时读/写数据，那就需要保证多个副本之间互相（N×N条通路）同步数据，数据的一致性和有序性非常难保证，大大增加了数据同步的复杂性，同时也增加了出现异常的几率。而引入Leader后，只有Leader负责数据读写，Follower只向Leader顺序pull数据（N条通路），系统更加简单且高效。
　　　　Kafka在ZooKeeper中动态维护了一个ISR（in-sync Replica），这个ISR里的所有副本都跟上了leader，只有ISR里的成员才有被选为Leader的可能。在这种模式下，对于f+1个副本，一个Partition能在保证不丢失已经commit的消息的前提下容忍f个副本的失败。在大多数使用场景中，这种模式是非常有利的。避免了所有副本参与选举，剔除过慢的副本。
　　　　还有一种非常常用的选举leader的方式是“Majority Vote”（“少数服从多数”），存在一定的缺点。但是数据保存更加准确。
　　3）副本均匀分布到整个集群
　　　　为了更好的做负载均衡，Kafka尽量将所有的Partition均匀分配到整个集群上。一个典型的部署方式是一个Topic的Partition数量大于Broker的数量。同时为了提高Kafka的容错能力，也需要将同一个Partition的副本尽量分散到不同的机器。实际上，如果所有的副本都在同一个Broker上，那一旦该Broker宕机，该Partition的所有副本都无法工作，也就达不到高可用的效果。同时，如果某个Broker宕机了，需要保证它上面的负载可以被均匀的分配到其它幸存的所有Broker上。
　　　　Kafka分配副本的算法如下：
　　　　　　1.将所有Broker（假设共n个Broker）和待分配的Partition排序
　　　　　　2.将第i个Partition分配到第（i % n 取模）个Broker上
　　　　　　3.将第i个Partition的第j个副本分配到第（(i + j) % n 取模）个Broker上
　　4）副本消息传递同步策略
　　　　Producer在发布消息到某个Partition时，先通过ZooKeeper找到该Partition的Leader，然后无论该Topic的副本数为多少，Producer只将该消息发送到该Partition的Leader。Leader会将该消息写入pagecache（由操作系统刷入log文件）。每个Follower都从Leader999pull数据。这种方式上，Follower存储的数据顺序与Leader保持一致。Follower在收到该消息写入pagecache（由操作系统刷入log文件），向Leader发送ACK。一旦Leader收到了ISR中的所有副本的ACK，该消息就被认为已经commit了，Leader将增加offset并且向Producer发送ACK。
　　　　为了提高性能，Kafka只能保证它被存于多个副本的内存中，而不能保证它们被持久化到磁盘中，一但所有副本进程消失，重启后，消息会被丢失。Consumer读消息也是从Leader读取，只有被commit过的消息才会暴露给Consumer。
　　　　Kafka Replication的数据流如下图所示：
　　　　

　　5）ACK前需要保证有多少个备份
　　　　对于Kafka而言，定义一个Broker是否“活着”包含两个条件：
　　　　　　一是它必须维护与ZooKeeper的session（这个通过ZooKeeper的Heartbeat机制来实现）。
　　　　　　二是Follower必须能够及时将Leader的消息复制过来，不能“落后太多”。
　　　　Leader会跟踪与其保持同步的副本列表（活跃副本数），该列表称为ISR（即in-syncReplica）。如果一个Follower宕机，或者落后太多，Leader将把它从ISR中移除，放入到OSR中（即out-syncReplica）。这里所描述的“落后太多”指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，这种复制方式极大的影响了吞吐率（取决于最慢的副本）。而异步复制方式下，Follower异步的从Leader复制数据，数据只要被Leader写入log就被认为已经commit，这种情况下缺点就是如果Follower都复制完都落后于Leader，而如果Leader突然宕机，则会丢失数据。
　　　　而Kafka的这种使用ISR的方式（活跃度高、复制足够快的副本同步完成），这条消息才会被认为commit，这样均衡了确保数据不丢失以及吞吐率。Follower可以批量的从Leader复制数据，这样极大的提高复制性能（批量写磁盘），极大减少了Follower与Leader的差距。
　　　　需要说明的是，Kafka只解决fail/recover，不处理“Byzantine”（“拜占庭”）问题。一条消息只有被ISR里的所有Follower都从Leader复制过去才会被认为已提交。这样就避免了部分数据被写进了Leader，还没来得及被任何Follower复制就宕机了，而造成数据丢失（Consumer无法消费这些数据）。而对于Producer而言，它可以选择是否等待消息commit，这可以通过request.required.acks来设置。这种机制确保了只要ISR有一个或以上的Follower，一条被commit的消息就不会丢失。
　　6）如何处理所有副本都不工作
　　　　在ISR中至少有一个follower时，Kafka可以确保已经commit的数据不丢失，但如果某个Partition的所有副本都宕机了，就无法保证数据不丢失了。这种情况下有两种可行的方案：
　　　　1.等待ISR中的任一个副本“活”过来，并且选它作为Leader
　　　　2.选择第一个“活”过来的副本（不一定是ISR中的）作为Leader
　　　　这就需要在可用性和一致性当中作出一个简单的折衷。如果一定要等待ISR中的副本“活”过来，那不可用的时间就可能会相对较长（既如果活过来的是OSR中的，还需要等待）。而且如果ISR中的所有副本都无法“活”过来了，或者数据都丢失了，这个Partition将永远不可用。选择第一个“活”过来的副本作为Leader，而这个副本不是ISR中的副本，那即使它并不保证已经包含了所有已commit的消息，它也会成为Leader而作为consumer的数据源。Kafka支持用户通过配置选择这两种方式中的一种，从而根据不同的使用场景选择高可用性还是强一致性。
　　7）选举Leader优化
　　　　最简单最直观的方案是，所有Follower都在ZooKeeper上设置一个Watch，一旦Leader宕机，其对应的临时节点会自动删除，此时所有Follower都尝试创建该节点，而创建成功者（ZooKeeper保证只有一个能创建成功）即是新的Leader，其它副本即为Follower。
　　　　但是该方法会有3个问题：
　　　　　　1.split-brain 这是由ZooKeeper的特性引起的，虽然ZooKeeper能保证所有Watch按顺序触发，但并不能保证同一时刻所有副本“看”到的状态是一样的，这就可能造成不同副本的响应不一致
　　　　　　2.herd effect 如果宕机的那个Broker上的Partition比较多，会造成多个Watch被触发，造成集群内大量的调整
　　　　　　3.ZooKeeper负载过重 每个副本都要为此在ZooKeeper上注册一个Watch，当集群规模增加到几千个Partition时ZooKeeper负载会过重。
　　　　Kafka 解决了上述问题，它在所有broker中选出一个controller，所有Partition的Leader选举都由controller决定。controller会将Leader的改变直接通过RPC的方式（比ZooKeeper Queue的方式更高效）通知需为为此作为响应的Broker。同时controller也负责增删Topic以及副本的重新分配。
二 、Kafka生产过程分析
　　1）写入方式
　　　　producer采用推（push）模式将消息发布到broker，每条消息都被追加（append）到分区（patition）中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障kafka吞吐率）。
　　2）分区（Partition）
　　　　Kafka集群有多个消息代理服务器（broker-server）组成，发布到Kafka集群的每条消息都有一个类别，用主题（topic）来表示。通常，不同应用产生不同类型的数据，可以设置不同的主题。一个主题一般会有多个消息的订阅者，当生产者发布消息到某个主题时，订阅了这个主题的消费者都可以接收到生成者写入的新消息。
　　　　Kafka集群为每个主题维护了分布式的分区（partition）日志文件，物理意义上可以把主题（topic）看作进行了分区的日志文件（partition log）。主题的每个分区都是一个有序的、不可变的记录序列，新的消息会不断追加到日志中。分区中的每条消息都会按照时间顺序分配到一个单调递增的顺序编号，叫做偏移量（offset），这个偏移量能够唯一地定位当前分区中的每一条消息。消息发送时都被发送到一个topic，其本质就是一个目录，而topic是由一些Partition Logs(分区日志)组成，其组织结构如下图所示：
　　　　

　　　　

　　　　图中的topic有3个分区，每个分区的偏移量都从0开始，不同分区之间的偏移量都是独立的，不会相互影响。
　　　　我们可以看到，每个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset值。发布到Kafka主题的每条消息包括键值和时间戳。消息到达服务器端的指定分区后，都会分配到一个自增的偏移量。原始的消息内容和分配的偏移量以及其他一些元数据信息最后都会存储到分区日志文件中。消息的键也可以不用设置，这种情况下消息会均衡地分布到不同的分区。
　　　　分区的原因：
　　　　（1）方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；
　　　　（2）可以提高并发，因为可以以Partition为单位读写了。
　　　　传统消息系统在服务端保持消息的顺序，如果有多个消费者消费同一个消息队列，服务端会以消费存储的顺序依次发送给消费者。但由于消息是异步发送给消费者的，消息到达消费者的顺序可能是无序的，这就意味着在并行消费时，传统消息系统无法很好地保证消息被顺序处理。虽然我们可以设置一个专用的消费者只消费一个队列，以此来解决消息顺序的问题，但是这就使得消费处理无法真正执行。
　　　　Kafka比传统消息系统有更强的顺序性保证，它使用主题的分区作为消息处理的并行单元。Kafka以分区作为最小的粒度，将每个分区分配给消费者组中不同的而且是唯一的消费者，并确保一个分区只属于一个消费者，即这个消费者就是这个分区的唯一读取线程。那么，只要分区的消息是有序的，消费者处理的消息顺序就有保证。每个主题有多个分区，不同的消费者处理不同的分区，所以Kafka不仅保证了消息的有序性，也做到了消费者的负载均衡。
　　　　分区的原则
　　　　（1）指定了patition，则直接使用；
　　　　（2）未指定patition但指定key，通过对key的value进行hash出一个patition
　　　　（3）patition和key都未指定，使用轮询选出一个patition。

DefaultPartitioner类

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (keyBytes == null) {
            int nextValue = nextValue(topic);
            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            if (availablePartitions.size() > 0) {
                int part = Utils.toPositive(nextValue) % availablePartitions.size();
                return availablePartitions.get(part).partition();
            } else {
                // no partitions are available, give a non-available partition
                return Utils.toPositive(nextValue) % numPartitions;
            }
        } else {
            // hash the keyBytes to choose a partition
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }

　　3）副本
　　　　同一个partition可能会有多个副本tion（对应 server.properties 配置中的 default.replication.factor=N）。
　　4） 写入流程
　　　　

　　　　producer写入消息流程如下：
　　　　1）producer先从zookeeper的 "/brokers/.../state"节点找到该partition的leader
　　　　2）producer将消息发送给该leader
　　　　3）leader将消息写入本地log
　　　　4）followers从leader pull消息，写入本地log后向leader发送ACK
　　　　5）leader收到所有ISR中的replication的ACK后，增加HW（high watermark，最后commit 的offset）并向producer发送ACK
三 、broker保存消息
　　1）存储方式
　　　　

　　　　物理上把 topic 分成一个或多个 patition（对应 server.properties 中的 num.partitions=3 配置），每个 patition 物理上对应一个文件夹（该文件夹存储该 patition 的所有消息和索引文件）。
　　2）存储策略
　　　　无论消息是否被消费，kafka 都会保留所有消息。有两种策略可以删除旧数据：

1、 基于时间：log.retention.hours=168 
2、 基于大小：log.retention.bytes=1073741824

　　3）Zookeeper存储结构
　　　　　　　　　　　　       　　admin
　　　　该目录下znode只有在有相关操作时才会存在，操作结束时会将其删除。/admin/reassign_partitions用于将一些Partition分配到不同的broker集合上。对于每个待重新分配的Partition，Kafka会在该znode上存储其所有的Replica和相应的Broker id。该znode由管理进程创建并且一旦重新分配成功它将会被自动移除。
　　　　broker
　　　　即/brokers/ids/[brokerId]）存储“活着”的broker信息。topic注册信息（/brokers/topics/[topic]），存储该topic的所有partition的所有replica所在的broker id，第一个replica即为preferred replica，对一个给定的partition，它在同一个broker上最多只有一个replica,因此broker id可作为replica id。
　　　　controller
　　　　/controller -> int (broker id of the controller)存储当前controller的信息
　　　　/controller_epoch -> int (epoch)直接以整数形式存储controller epoch，而非像其它znode一样以JSON字符串形式存储。
四 、Kafka消费过程分析
　　1）消费模型
　　　　消息的消费模型有两种：推送模型（push）和拉取模型（pull）。
　　　　基于推送模型（push）的消息系统，由消息代理记录消费者的消费状态。消息代理在将消息推送到消费者后，标记这条消息为已消费，但这种方式无法很好地保证消息被处理。比如，消息代理把消息发送出去后，当消费进程挂掉或者由于网络原因没有收到这条消息时，就有可能造成消息丢失（因为消息代理已经把这条消息标记为已消费了，但实际上这条消息并没有被实际处理）。如果要保证消息被处理，消息代理发送完消息后，要设置状态为“已发送”，只有收到消费者的确认请求后才更新为“已消费”，这就需要消息代理中记录所有的消费状态，这种做法显然是不可取的。
　　　　Kafka采用拉取模型，由消费者自己记录消费状态，每个消费者互相独立地顺序读取每个分区的消息。如下图所示，有两个消费者（不同消费者组）拉取同一个主题的消息，消费者A的消费进度是3，消费者B的消费进度是6。消费者拉取的最大上限通过最高水位（watermark）控制，生产者最新写入的消息如果还没有达到备份数量，对消费者是不可见的。这种由消费者控制偏移量的优点是：消费者可以按照任意的顺序消费消息。比如，消费者可以重置到旧的偏移量，重新处理之前已经消费过的消息；或者直接跳到最近的位置，从当前的时刻开始消费。
　　　　在一些消息系统中，消息代理会在消息被消费之后立即删除消息。如果有不同类型的消费者订阅同一个主题，消息代理可能需要冗余地存储同一消息；或者等所有消费者都消费完才删除，这就需要消息代理跟踪每个消费者的消费状态，这种设计很大程度上限制了消息系统的整体吞吐量和处理延迟。Kafka的做法是生产者发布的所有消息会一致保存在Kafka集群中，不管消息有没有被消费。用户可以通过设置保留时间来清理过期的数据，比如，设置保留策略为两天。那么，在消息发布之后，它可以被不同的消费者消费，在两天之后，过期的消息就会自动清理掉。
　　　　pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直等待数据到达。为了避免这种情况，我们在我们的拉请求中有参数，允许消费者请求在等待数据到达的“长轮询”中进行阻塞（并且可选地等待到给定的字节数，以确保大的传输大小）。
　　2）API 使用
　　　　既可以不需要自行去管理offset，系统通过zookeeper自行管理。不需要管理分区，副本等情况，系统自动管理。消费者断线会自动根据上一次记录在zookeeper中的offset去接着获取数据（默认设置1分钟更新一下zookeeper中存的offset）
　　　　也可以开发者自己控制offset，想从哪里读取就从哪里读取。自行控制连接分区，对分区自定义进行负载均衡对zookeeper的依赖性降低（如：offset不一定非要靠zk存储，自行存储offset即可，比如存在文件或者内存中）
　　3）消费者组
　　　　

　　　　消费者是以consumer group消费者组的方式工作，由一个或者多个消费者组成一个组，共同消费一个topic。每个分区在同一时间只能由group中的一个消费者读取，但是多个group可以同时消费这个partition。在图中，有一个由三个消费者组成的group，有一个消费者读取主题中的两个分区，另外两个分别读取一个分区。某个消费者读取某个分区，也可以叫做某个消费者是某个分区的拥有者。
　　　　在这种情况下，消费者可以通过水平扩展的方式同时读取大量的消息。另外，如果一个消费者失败了，那么其他的group成员会自动负载均衡读取之前失败的消费者读取的分区。
五 、Topic的创建和删除
　　1）创建topic
　　　　创建 topic 的序列图如下所示：
　　　　

流程说明：
1、 controller 在 ZooKeeper 的 /brokers/topics 节点上注册 watcher，当 topic 被创建，则 controller 会通过 watch 得到该 topic 的 partition/replica 分配。
2、 controller从 /brokers/ids 读取当前所有可用的 broker 列表，对于 set_p 中的每一个 partition：
     2.1、 从分配给该 partition 的所有 replica（称为AR）中任选一个可用的 broker 作为新的 leader，并将AR设置为新的 ISR 
     2.2、 将新的 leader 和 ISR 写入 /brokers/topics/[topic]/partitions/[partition]/state 
3、 controller 通过 RPC 向相关的 broker 发送 LeaderAndISRRequest。

　　2 ）删除topic
　　　　删除 topic 的序列图如下所示：
　　　　

流程说明：
1、 controller 在 zooKeeper 的 /brokers/topics 节点上注册 watcher，当 topic 被删除，则 controller 会通过 watch 
得到该 topic 的 partition/replica 分配。 
2、 若 delete.topic.enable=false，结束；否则 controller 注册在 /admin/delete_topics 上的 watch 被 fire，controller 
通过回调向对应的 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.enable=true（默认值是 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.enable=true（默认值是true），则启动 partition-rebalance 线程。 
12、 若 delete.topic.enable=true 且Delete Topic Patch(/admin/delete_topics)中有值，则删除相应的Topic。