标签：影响 leader term 数据表 调用 while sock 用处 ati
在上篇中我们了解了Kafka是什么，为什么需要Kafka，以及Kafka的基本架构和各自的作用是什么，这篇文章中我们将从kafka内部每一个组成部分去看kafka 是如何保证数据的可靠性以及工作机制。因为时间问题，或许排版多有瑕疵，有些内容未能做到详尽。待之后有空会前来填坑。话不多说，正片开始：
Kafka中消息是以topic进行分类的，生产者生产消息，消费者消费消息，都是面向topic的。
topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，以便出错恢复时，从上次的位置继续消费。
为什么分区？
1）方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；
2）可以提高并发，因为可以以Partition为单位读写了。
Kafka 使用 Zookeeper 来维护集群成员的信息。每个 broker （每一个节点就是一个broker）都有一个唯一标识符，这个标识符可以自动生成，也可以在配置文件里指定(我们一般也这样做，常见的做法是通过kafka安装目录下conf/server.properties 文件进行配置) 。配置如下：
# see kafka.server.KafkaConfig for additional details and defaults
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
# 这个id值 集群全局唯一
broker.id=2
在 broker 启动的时候，它通过创建临时节点把自己的 ID 注册到 Zookeeper， Kafka 组件订阅 Zookeeper 的／brokers/ids 路径 (broker在Zookeeper 上的注册路径），当有 broker 加入集群或退出集群时，这些组件就可以获得通知。
在broker 停机、出现网络分区或长时间垃圾回收停顿时，broker 会从 Zookeeper 上断开连 接，此时 broker 在启动时创建的临时节点会自动从 Zookeeper 上移除。监听 broker 列表的 Kafka 组件会被告知该 broker 已移除。
当集群启动之后，Kafka集群开始工作了，如上图所示：
首先，集群启动后，集群中的broker会通过选举机制选出一个控制器Controller，具体的选举细节在后边在进行细说。控制器除了具有一般的broker的功能之处，还负责分区leader的选举。我们到现在已经知道：
kafka使用主题Topic来进行组织数据，
每个主题被分成若干个分区（分区一般在我们创建topic的时候指定，默认是1个分区）；
每个分区有多个副本(副本数量一般同样是我们创建的时候指定，默认为1 ，但是其值不能超过节点的个数，因为副本是均衡分布的)。
生产者会创建一个ProducerRecord对象通过指定的主题向集群发送消息，ProducerRecord对象需要将消息的键值序列化才能在网络中传输。数据被发送到集群中的某个broker的时候，这个时候会先经过分区器确认数据要写入在那个分区。这时候分区器对于数据的键key进行检测，会有如下三种情况：
确定好分区信息后，生产者就知道该往那个主题和那个分区发送该条记录了。但是这个消息不会立即发送，而是将这条记录添加到一个记录批次中，这个批次的所有消息都是发送同一主题和分区的。批次发送有两个参数：设定的时间和批次的容量，只要满足其一就发送。发送是由一个独立的线程负责处理。服务器也就是broker收到消息返回是否写入成功，
当对应主题和分区的broker接收到一批数据写入请求时，broker先进行一些验证：
我们知道kafka是分布式消息队列，如何保证数据的可靠性和不丢失是重中之重。生产者写入过程(Kafka 还从broker内部也就是分区的方面进行了可靠性的保障机制，稍后展开)如何进行可靠性的保证，Kafka采用了Acks应答机制。topic的每个partition收到producer发送的数据后，都需要向producer发送ack（acknowledgement确认收到），如果producer收到ack，就会进行下一轮的发送，否则重新发送数据。
Acks机制提供了三种可靠性级别：
这时候，broker开始写入producer发送的一批数据。Kafka是顺序写磁盘的方式持久化数据，在我们的认知中，是不是觉得写磁盘很慢，但有大量数据请求写入，怕是写的黄花菜都凉了哦。别急，kafka 能够如此火热自然有其特殊之处，正所谓：没有金刚钻，不揽瓷器活嘛。Kafka是对于数据进行追加的方式顺序写入，这样就减少了大量的磁头寻址的时间。官网数据表明：同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。
由于生产者生产的消息会不断追加到log文件末尾，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment对应三个文件——“.index”文件、“.timeindex”文件和“.log”文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。
segment的命名规则:
1、每个分区第一个segment的文件名= 0000000000000000000
2、后续第N个segment文件名 = 第N-1个segment中最后一个offset+1
segment给log文件建索引的时候是每个一段范围[4k]建一个索引，是为稀疏索引。
“.index”文件存储大量的索引信息，“.log”文件存储大量的数据，索引文件中的元数据指向对应数据文件中message的物理偏移地址。如果我们现在查询第三条数据即offset=2的数据：则其索引为000000000000000002，通过在index文件中确认其索引，找到对应的数据记录的log中的地址，然后再找到对应的数据位置，读取出来。
producer向leader写入数据之后，follower需要向leader同步消息信息也就是需要复制多少份数据，但是我们应该配置多少个副本呢？又因为副本的均衡分布，就是一个broker只会有同主题同分区的一个副本。那配置副本就是配置broker，也就是需要多少个节点可以满足我们数据的可靠性保证呢？在Kafka中，每个分区的默认副本数=3。就是说最小集群的配置数，HDFS的默认副本也是3，所以一般3副本就足以保证数据不会丢失，当然也要考虑机架配置的哦。如果复制系数为N ，那么在N-1个broker 失效的情况下，仍然能够从主题读取数据或向主题写入数据。所以，更高的复制系数会带来更高的可用性、可靠性和更少的故障。另一方面，复制系数N需要至少N个broker ，而且会有N个数据副本。我们可以根据自身需求来确认，比如：银行为了保证数据更高的可靠性，就可以将复制系数设置为5。如果我们可以接受主题偶尔的不可用，也可以配置为2，当一台broker崩溃，另一台broker作为新的controller继续进行后续的工作。
在前边我们了解Ack应答机制有三种应答级别。最为可靠的设置为all。在当前应答级别下，假设leader收到数据，所有follower都开始同步数据，但有一个follower，因为某种故障，迟迟不能与leader进行同步，那leader就要一直等下去，直到它完成同步，才能发送ack。这个问题怎么解决呢？
Leader维护了一个动态的in-sync replica set (ISR)，意为和leader保持同步的follower集合。当ISR中的follower完成数据的同步之后，leader就会给producer发送ack。如果follower长时间未向leader同步数据，则该follower将被踢出ISR，该时间阈值由replica.lag.time.max.ms参数设定。Leader发生故障之后，就会从ISR中选举新的leader。ISR队列中follower的选择标准：
我们知道消费者也是只和leader分区进行通讯进行消费数据，如下：
1）follower故障
follower发生故障后会被临时踢出ISR，待该follower恢复后，follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向leader进行同步。等该follower的LEO大于等于该Partition的HW，即follower追上leader之后，就可以重新加入ISR了。
2）leader故障
leader发生故障之后，会从ISR中选出一个新的leader，之后，为保证多个副本之间的数据一致性，其余的follower会先将各自的log文件高于HW的部分截掉，然后从新的leader同步数据。
注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。
Kafka消费者从属于消费者群组。一个群组里的消费者订阅的是同一个主题，每个消费者接收主题一部分分区的消息。假设主题T1有4个分区，我们创建了消费者C1，他是群组G1中唯一的消费者，我们用它订阅主题T1，消费者C1将收到主题T1全部4个分区的消息:
如果群组G1新增一个消费者C2，那么每个消费者将分别从两个分区接受消息。我们假设消费者C1消费分区0和分区2的消息，消费者C2接收消费分区1和分区3的消息，如图：
如果群组G1有4个消费者，那么每个消费者都分配到一个分区：
如果我们继续往群组里添加更多消费者，超过主题的分区数量，那么有一部分消费者就会被闲置，不会接收到任何的消息：
往群组里增加消费者是横向伸缩消费能力的主要方式。 Kafka 消费者经常会做一些高延迟 的操作，比如把数据写到数据库或 HDFS ，或者使用数据进行比较耗时的计算。在这些情况下，单个消费者无法跟上数据生成的速度，所以可以增加更多的消费者，让它们分担负载，每个消费者只处理部分分区的消息，这就是横向伸缩的主要手段。我们有必要为主题创建大量的分区，在负载增长时可以加入更多的消费者。不过要注意，不要让消费者的数量超过主题分区的数量，多余的消费者只会被闲置。
除了通过增加消费者来横向伸缩单个应用程序外，还经常出现多个应用程序从同 主题 读取数据的情况。实际上， Kafka 设计的主要目标之 ，就是要让 Kafka 主题里的数据能 够满足企业各种应用场景的需求。在这些场景里，每个应用程序可以获取到所有的消息， 而不只是其中的部分。只要保证每个应用程序有自己的消费者群组，就可以让它们获取到主题所有的消息。不同于传统的消息系统，横向伸缩 Kafka 消费者和消费者群组并不 对性能造成负面影响。
在上面的例子里，如果新增 个只包含 个消费者的群组 G2 ，那么这个消费者将从主题 Tl 上接收所有的消息，与群组 Gl 之间互不影响。群组 G2 可以增加更多的消费者，每个 消费者可以悄费若干个分区，就像群组 Gl 那样，如图所示。总的来说，群组 G2 还是 会接收到所有消息，不管有没有其他群组存在。
我们通过上边的例子知道，群组里的消费者共同读取主题的分区。一个新的悄费者加 入群组时，它读取的是原本由其他消费者读取的消息。当一个消费者被关闭或发生崩愤时，它就离开群组，原本由它读取的分区将由群组里的其他消费者来读取。在主题发生变化时 比如管理员添加了新的分区，会发生分区重分配。 分区的所有权从 个消费者转移到另 个消费者，这样的行为被称为再均衡。再均衡非常 重要， 它为消费者群组带来了高可用性和伸缩性（我们可以放心地添加或移除梢费者）， 不过在正常情况下，我们并不希望发生这样的行为。在再均衡期间，消费者无法读取消息，造成整个群组一小段时间的不可用。另外，当分区被重新分配给另一个消费者时，消费者当前的读取状态会丢失，它有可能还需要去刷新缓存，在它重新恢复状态之前会拖慢应用程序。如何进行安全的再均衡，以及如何避免不必要的再均衡。
消费者通过向被指派为群组协调器的 broker （不同的群组可以有不同的协调器）发送心跳来维持它们和群组的从属关系以及它们对分区的所有权关系。只要消费者以正常的时间间隔发送心跳，就被认为是活跃的，说明它还在读取分区里的消息。消费者会在轮询消息 （为了获取消息）或提交偏移量时发送心跳。如果消费者停止发送心跳的时间足够长，会话就会过期，群组协调器认为它已经死亡，就会触发一次再均衡。 如果一个消费者发生崩溃，井停止读取消息，群组协调器会等待几秒钟，确认它死亡了才 触发再均衡。在这几秒钟时间里，死掉的消费者不会读取分区里的消息。在清理消费者时，消费者会通知协调器它将要离开群组，协调器会立即触发一次再均衡，尽量降低处理停顿。
当消费者要加入群组时，它会向群组协调器发送 Join Group 请求。第1个加入群组的消费者将成为“群主”。群主从协调器那里获得群组的成员列表（列表中包含了所有最近发送过心跳的消费者，它们被认为是活跃的），并负责给每一个消费者分配分区。它使用 个实现了 PartitionAssignor接口的类来决定哪些分区应该被分配给哪个消费者。分配完毕之后，群主把分配情况列表发送给群组协调器，协调器再把这些信息发 送给所有消费者。每个消费者只能看到自己的分配信息，只有群主知道群组里所有消费者的分配信息。这个过程会在每次再均衡时重复发生。
上边我们知道，分区会被分配个群组里的消费者。PartitionAssignor 根据给定的消费者和主题，决定哪些分区应该被分配给哪个消费者。Kafka 有两个默认的分配策：Range 和RoundRobin。
Range：
该策略会把主题的若干个连续的分区分配给消费者。假设消费者C1和消费者 C2 同时 订阅了主题 T1 和主题 T2 ，井且每个主题有3个分区。那么消费者 C1有可能分配到这两个主题的分区1和分区3，而消费者 C2 分配到这两个主题的分区2 。因为每个主题拥有奇数个分区，而分配是在主题内独立完成的，第一个消费者最后分配到比第二个消费者更多的分区。只要使用了Range策略，而且分区数量无法被消费者数量整除，就会出现这种情况。
如图中，第一次先分配Topic1的三个分区，对于消费者，就是C1先分第一块，然后C2分得第二块，C1接着分配到第三块。同样对于第二个Topic，同样的顺序进行分配分区，C1分得第一块分区，C2分得第二块分区，C1接着分得第三块。最后就是C1分配到4个分区进行消费，而C2只分得两个分区。
RoundRobin
该策略把主题的所有分区逐个分配给消费者。如果使用 RoundRobin 策略来给消费者 C1和消费者 C2 分配分区，那么消费者C1将分到主题 T1的分区1和分区3以及主题 T2 的分区2 ，消费者 C2 将分配到主题 T1分区2 以及主题T2的分区1和分区3。一般 来说 ，如果所有消费者都订阅相同的主题(这种情况很常见), RoundRobin 策略会给所有消费者分配相同数量的分区（或最多就差1个分区）。
如图：RoundRobin策略相当于把当前消费者组订阅的主题中所有分区看做统一的整体，然后对消费者群组中的每一个活跃者进行轮流分配。
在此，我们有必要明白Consumer是如何消费的。调用了那些方法，做了那些行为来完成一次消费；
对于轮询阶段我们进行详细分析说明：
轮询不只是获取数据那么简单。在第一次调用新消费者的 poll()方法时，它会负责查找 GroupCoordinator 然后加入群组，接受分配的分区。如果发生了再均衡，整个过程也 在轮询期间进行 。当然，心跳也是从轮询里发送出去的。所以，我们要确保在轮询期间，所做的任何处理工作都应该尽快完成。
每次调用 poll ()方法，它总是返回由生产者写入 Kafka 但还没有被消费者读取过的记录 我们因此可以追踪到哪些记录是被群组里的哪个消费者读取的。这是 Kafka 个独特之处。消费者可以使用 Kafka 来追踪消息在分区里的位置（偏移量）。
我们把更新分区当前位置的操作叫作提交。消费者消费消息是按照批次进行的。
那么消费者是如何提交偏移量的呢？消费者往一个叫作 __consumer_offset 特殊主题发送消息，消息里包含每个分区的偏移量。如果消费者一直处于运行状态，那么偏移量就没有什么用处。不过，如果悄费者发生崩溃或者有新的消费者加入群组，就会触发再均衡，完成再均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。说人话也就是说消费者群组发生变化的时候或者消费者组重启之后，要能从上一次消费的地方接着消费数据。消费者组会知道并记录每一次消费的时候消费者消费主题分区的最后一个消息的offset，这样，下一次当前消费者组再开始消费的时候，就能从具体分区的最后一次消费的地方接着消费。
如果提交的偏移量小于客户端处理的最后一个消息的偏移量 ，那么处于两个偏移量之间的消息就会被重复处理，如图：
如果提交的偏移量大于客户端处理的最后 个消息的偏移量，那么处于两个偏移量之间的 消息将会丢失:
所以，处理偏移量的方式对客户端会有很大的影响。
最简单的提交方式是让悄费者自动提交偏移量。如果 enable .auto.commit 被设为 true ，那 么每过5s，消费者会自动把从 poll()方法接收到的最大偏移量提交上去。提交时间间隔 auto.commit.interval.ms 控制，默认值是 5s 。与梢费者里的其他东西一样，自动提交也是在轮询里进行的。消费者每次在进行轮询时会检查是否该提交偏移量了，如果是，那么就会提交从上一次轮询返回的偏移量。
不过当前策略有什么缺陷呢？可以想想。
假设我们仍然使用默认的 5s 提交时间间隔，在最近一次提交之后的 3s 发生了再均衡，再均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后 3s ，所以在这 3s 内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复悄息的时间窗，不过这种情况是无法完全避免的。
在使用自动提交 ，每次调用轮询方法上一次调用返回的偏移量提交上去，它并不知道具体哪些消息已经被处理了，所以在再次调用之前最好确保所有当前调用返回的消息都已经处理完毕（在调用 close()方位之前也 行自动提交）。
我们可以通过控制提交偏移量的时间尽可能消除丢失消息的可能性和再均衡时重复消费数据的数量。此外消费者API 提供了另一种提交偏移量的方式 ，让我们可以基于处理消息的时候需要提交的去提交当前偏移盘，而不是基于时间间隔。
首先我们需要在消费者的配置中关闭自动提交参数：auto.commit.offset 设为false，让应用程序决定何时提交偏移量。
使用 commitSync() 提交偏移量最简单最可靠，因为这个方法是同步方法。这个方法会提交由 poll()方法返回的最新偏移量，提交成功后马上返回，如果提交失败就抛出异常。
要记住， commitSync() 将会提交由 poll ()返回的最新偏移量,所以在处理完所有记录后要确保调用了 commitSync() ，否则还是会有丢失消息的风险。如果发生了再均衡，从最近一批消息到发生再均衡之间的所有消息都将被重复处理。
同步提交有一个不足之处， 在broker对提交请求作出回应之前，应用程序会一直阻塞，这样会限制应用程序的吞吐量。我们可以通过降低提交频率来提升吞吐量，但如果发生了再均衡， 会增加重复消息的数量。这时候我们可以使用异步提交方式进行提交：commitASync()。
这时候我们只发送提交请求，不用等待broker的响应.
我们从kafka 的工作机制，从生产者 到broker集群到消费者全套的工作流程和内部的细节都有一个比较清晰的认知了。今天匆匆就结束这个篇章，待之后再来细化排布。
怕什么真理无穷，进一寸有一寸的欢喜。我是清风，希望这篇文章对你有帮助。如有不准确之处，还请评论区留言讨论。
标签：影响 leader term 数据表 调用 while sock 用处 ati
原文地址：https://www.cnblogs.com/liangzilx/p/14868874.html
