kafka最佳实践

Posted 2023-01-02 Proxbj

注意事项

1. kafka重平衡比较坑，当客户端收不到最新的消息时，大概率是kafka在重平衡，可以查看消费位点，查看kafka是否活跃，是否在重平衡。
   

常用命令

1. docker中查看消费位点:

docker exec -it kafka /bin/bash

cd /opt/kafka/bin

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group dev

2. docker中修改kafka的消费位点

• 如果重置,需要将使用group的程序停掉

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group dev --reset-offsets --topic dev --to-earliest --execute

一、概念

1. 简介

• Apache Kafka 是一款开源的消息引擎系统，也是一个分布式流处理平台
• 消息使用纯二进制的字节序列传输
• 消息消费分为点对点模型和发布/订阅模型

2. 术语

• 消费者、生产者不多说，都被称为客户端
• Broker。kafka的服务器端被称为Broker的服务进程构成，即一个服务器端由多个Broker构成。Broker负责接收和处理客户端发送过来的请求，以及对消息进行持久化。虽然多个Broker可以分布在同一台机器上，但是一般部署在多台机器，能够做到高可用。
• 主题。主题就是topic，理解为语义相近的消息容器，发布/订阅模型主要面向的就是topic。
• 分区。如果一个Broker上面保存的消息太多了，是不是应该考虑将消息保存到多个Broker上面，这样才能够保证伸缩性。kafka就是依靠将一个topic打散分为多个分区，每个分区是一组有序的消息日志。生产者只会发送一条消息到一个分区中，分配的规则由kafka决定。消费者会均分一个topic中的所有分区。
• 副本。虽然Broker分到了多个机器上面，也只能保证消息处理的高可用，如果某台机器宕机了，消息也丢失了找不回来了，所以需要备份机制。kafka使用副本保证备份，一个分区会分为多个副本，每个分区都有一个领导者副本和多个追随者副本。这些副本会分散在多台机器上，领导者副本对客户端提供读写，追随者副本只同步领导者副本的数据，不进行读写操作。
• 消息持久化。总的来说，Kafka 使用消息日志（Log）来保存数据，一个日志就是磁盘上一个只能追加写（Append-only）消息的物理文件。因为只能追加写入，故避免了缓慢的随机 I/O 操作，改为性能较好的顺序 I/O 写操作，这也是实现 Kafka 高吞吐量特性的一个重要手段。不过如果你不停地向一个日志写入消息，最终也会耗尽所有的磁盘空间，因此 Kafka 必然要定期地删除消息以回收磁盘。怎么删除呢？简单来说就是通过日志段（Log Segment）机制。在 Kafka 底层，一个日志又进一步细分成多个日志段，消息被追加写到当前最新的日志段中，当写满了一个日志段后，Kafka 会自动切分出一个新的日志段，并将老的日志段封存起来。Kafka 在后台还有定时任务会定期地检查老的日志段是否能够被删除，从而实现回收磁盘空间的目的。
• 重平衡。消费者组里面的所有消费者实例不仅“瓜分”订阅主题的数据，而且更酷的是它们还能彼此协助。假设组内某个实例挂掉了，Kafka 能够自动检测到，然后把这个 Failed 实例之前负责的分区转移给其他活着的消费者。这个过程就是 Kafka 中大名鼎鼎的“重平衡”（Rebalance）。

二、生产者消息分区机制

1. 分区保存策略

• 轮询策略
• 随机策略
• 按消息键保存策略

三、无消息丢失

配置:

1. 不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，一定要使用带有回调通知的 send 方法。
2. 设置 acks = all。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。如果设置成 all，则表明所有副本 Broker 都要接收到消息，该消息才算是“已提交”。这是最高等级的“已提交”定义。
3. 设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。
4. 设置 unclean.leader.election.enable = false。这是 Broker 端的参数，它控制的是哪些 Broker 有资格竞选分区的 Leader。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，必然会造成消息的丢失。故一般都要将该参数设置成 false，即不允许这种情况的发生。
5. 设置 replication.factor >= 3。这也是 Broker 端的参数。其实这里想表述的是，最好将消息多保存几份，毕竟目前防止消息丢失的主要机制就是冗余。
6. 设置 min.insync.replicas > 1。这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1。
7. 确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。
8. 确保消息消费完成再提交。Consumer 端有个参数 enable.auto.commit，最好把它设置成 false，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的。

四、拦截器

1. 消费者拦截器

• 具体的实现类要实现 org.apache.kafka.clients.consumer.ConsumerInterceptor 接口，这里面也有两个核心方法。
• onConsume：该方法在消息返回给 Consumer 程序之前调用。也就是说在开始正式处理消息之前，拦截器会先拦一道，搞一些事情，之后再返回给你。
• onCommit：Consumer 在提交位移之后调用该方法。通常你可以在该方法中做一些记账类的动作，比如打日志等。

2. 生产者拦截器

• 所有 Producer 端拦截器实现类都要继承 org.apache.kafka.clients.producer.ProducerInterceptor 接口。该接口是 Kafka 提供的，里面有两个核心的方法。
• onSend：该方法会在消息发送之前被调用。如果你想在发送之前对消息“美美容”，这个方法是你唯一的机会。
• onAcknowledgement：该方法会在消息成功提交或发送失败之后被调用。onAcknowledgement 的调用要早于 callback 的调用。值得注意的是，这个方法和 onSend 不是在同一个线程中被调用的，因此如果你在这两个方法中调用了某个共享可变对象，一定要保证线程安全哦。还有一点很重要，这个方法处在 Producer 发送的主路径中，所以最好别放一些太重的逻辑进去，否则你会发现你的 Producer TPS 直线下降。

五、Producer管理tcp连接

1. KafkaProducer 实例创建时启动 Sender 线程，从而创建与 bootstrap.servers 中所有 Broker 的 TCP 连接。
2. KafkaProducer 实例首次更新元数据信息之后，还会再次创建与集群中所有 Broker 的 TCP 连接。
3. 如果 Producer 端发送消息到某台 Broker 时发现没有与该 Broker 的 TCP 连接，那么也会立即创建连接。
4. 如果设置 Producer 端 connections.max.idle.ms 参数大于 0，则步骤 1 中创建的 TCP 连接会被自动关闭；如果设置该参数 =-1，那么步骤 1 中创建的 TCP 连接将无法被关闭，从而成为“僵尸”连接。

六、幂等性和事务性

1.幂等

1. 背景

所谓的消息交付可靠性保障，是指 Kafka 对 Producer 和 Consumer 要处理的消息提供什么样的承诺。常见的承诺有以下三种：

• 最多一次（at most once）：消息可能会丢失，但绝不会被重复发送。
• 至少一次（at least once）：消息不会丢失，但有可能被重复发送。
• 精确一次（exactly once）：消息不会丢失，也不会被重复发送。
  Kafka 默认提供的交付可靠性保障是即至少一次，因为kafka 的producer 在消息发送失败(没有接收到kafka broker 的ACK信息)的时候则会进行重试，这就是kafka 为什么默认提供的是至少一次的交付语义，但是这样可能导致消息重复
  Kafka 也可以提供最多一次交付保障，只需要让 Producer 禁止重试即可。这样一来，消息要么写入成功，要么写入失败，但绝不会重复发送。我们通常不会希望出现消息丢失的情况，但一些场景里偶发的消息丢失其实是被允许的，相反，消息重复是绝对要避免的。此时，使用最多一次交付保障就是最恰当的。
  无论是至少一次还是最多一次，都不如精确一次来得有吸引力。大部分用户还是希望消息只会被交付一次，这样的话，消息既不会丢失，也不会被重复处理。或者说，即使 Producer 端重复发送了相同的消息，Broker 端也能做到自动去重。在下游 Consumer 看来，消息依然只有一条，而这就是我们今天要介绍的幂等性。

2. producer达成幂等性很简单，只需要配置一个属性就行

props.put(“enable.idempotence”, ture)
// 或 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG， true)

3. producer的幂等性需要注意以下事项：

• 只能保证单个分区幂等，不同分区幂等是不能保证的
• 只能保证单个会话幂等，也就是producer连接broker的单次会话才能保证幂等，不能跨会话

那么问题来了，怎么保证多个分区和跨会话的消息幂等呢？答案就是事务（transaction）或者依赖事务型 Producer。这也是幂等性 Producer 和事务型 Producer 的最大区别！

2.事务型 Producer

• 事务型 Producer 能够保证将消息原子性地写入到多个分区中。这批消息要么全部写入成功，要么全部失败。另外，事务型 Producer 也不惧进程的重启。Producer 重启回来后，Kafka 依然保证它们发送消息的精确一次处理。
• 设置事务型 Producer 的方法也很简单，满足两个要求即可：和幂等性 Producer 一样，开启 enable.idempotence = true。设置 Producer 端参数 transactional. id。最好为其设置一个有意义的名字。此外，你还需要在 Producer 代码中做一些调整，如这段代码所示：

producer.initTransactions();
try 
	producer.beginTransaction(); producer.send(record1); 
	producer.send(record2); 
	producer.commitTransaction();
	 
catch (KafkaException e)  
	producer.abortTransaction();
	

和普通 Producer 代码相比，事务型 Producer 的显著特点是调用了一些事务 API，如 initTransaction、beginTransaction、commitTransaction 和 abortTransaction，它们分别对应事务的初始化、事务开始、事务提交以及事务终止。
这段代码能够保证 Record1 和 Record2 被当作一个事务统一提交到 Kafka，要么它们全部提交成功，要么全部写入失败。
实际上即使写入失败，Kafka 也会把它们写入到底层的日志中，也就是说 Consumer 还是会看到这些消息。因此在 Consumer 端，读取事务型 Producer 发送的消息也是需要一些变更的。修改起来也很简单，设置 isolation.level 参数的值即可。当前这个参数有两个取值：

• read_uncommitted：这是默认值，表明 Consumer 能够读取到 Kafka 写入的任何消息，不论事务型 Producer 提交事务还是终止事务，其写入的消息都可以读取。
• read_committed：表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了，它也能看到非事务型 Producer 写入的所有消息。

七、消费者组

1. 消费者组特性：

• 消费者组里面的消费者实例共享topic的分区，并且一个分区只能分配给相同消费组下的一个消费者实例，当然该分区可以分配给其他消费者组的实例。
• 消费者组下可以有多个消费者实例
• 消费者组是一串字符串标识

2. 消息引擎一般是两种消费模型：点对点的队列消费模型，发布/订阅模型。点对点的模型：生产者和消费者是一对一的关系，就像打电话一样。发布/订阅模型：多个生产者生产消息到topic，多个消费者订阅topic的消息。

Kafka 仅仅使用 Consumer Group 这一种机制，却同时实现了传统消息引擎系统的两大模型：

• 如果所有实例都属于同一个 Group，那么它实现的就是消息队列模型；
• 如果所有实例分别属于不同的 Group，那么它实现的就是发布 / 订阅模型。

3. 消费者组使用注意。

• 消费者组中的实例个数，最好与订阅主题的分区数相同，否则多出的实例只会被闲置。一个分区只能被一个消费者实例订阅。
• 老版本的consumer的消费位点offset保存在zookeeper中，但是这样会频繁对zookeeper操作，影响zookeeper的性能。所以新版本，消费位点offset保存在kafka集群中。

4. 消费重平衡。

• 发生重平衡的条件：
  a，组成员数发生变更
  b，订阅主题数发生变更
  c，定阅主题分区数发生变更
• 影响：
  Rebalance 的设计是要求所有consumer实例共同参与，全部重新分配所有用分区。并且Rebalance的过程比较缓慢，这个过程消息消费会中止。

八、位移

1.查看消息位移

kafka管理消费位移，其实也是将消费的offset保存在一个topic里面，这个topic的名字是__consumer_offsets，当然不能向这个topic发消息，格式是有一定规定的。分区数依赖于Broker端的offsets.topic.num.partitions的取值，默认为50；副本数依赖于Broker端的offsets.topic.replication.factor的取值，默认为3。下图可以发现（部分分区截图）分区里面的消息记录还是比较多的。

该topic的消息格式为：

• key: __consumer_offsets
• value: 保存一些内容

1. 可以使用命令行查看该topic的消息：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic __consumer_offsets --partition 48 --from-beginning --formatter 'kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter'

• 由上图可以发现，在__consumer_offsets这个topic的48分区，保存了一些topic的消费位点

2. 当然也可以写代码查看该topic消息，只不过这个消息需要解码，具体就不演示了
   

2.提交位移

kafka支持两种位移提交的方式：自动提交和手动提交

1. 自动提交

• kafka自动提交将enable.auto.commit设置为true
• 并且将时间设置合理范围auto.commit.interval.ms
• 自动提交很方便，但是丧失了很大的灵活性和可控性，你完全没法把控 Consumer 端的位移管理

2. 手动提交

现在大数据框架一般都是禁用了自动提交，仅支持手动提交

• 手动提交将enable.auto.commit设置为false
• 在消费端使用consumer.commitSync ， 同步方式提交
• 在消费端使用consumer.commitAsync ， 同步方式提交

3. 删除过期位移消息

如果开启了自动提交，且提交时间为1s。则在1s后都会提交位移，不管这个位移是否发生改变。举个例子，在topic：test里面的位移为100，现在没有新消息产生和消费，则会间隔1s提交offset=100，这样一直累计下去，将会把kafka磁盘打爆。所以就需要删除过期消息，保存最新的消费位点消息。

• Kafka 是怎么删除位移主题中的过期消息的呢？答案就是 Compaction。国内很多文献都将其翻译成压缩，我个人是有一点保留意见的。在英语中，压缩的专有术语是 Compression，它的原理和 Compaction 很不相同，我更倾向于翻译成压实，或干脆采用 JVM 垃圾回收中的术语：整理。不管怎么翻译，Kafka 使用 Compact 策略来删除位移主题中的过期消息，避免该主题无限期膨胀。

• 那么应该如何定义 Compact 策略中的过期呢？对于同一个 Key 的两条消息 M1 和 M2，如果 M1 的发送时间早于 M2，那么 M1 就是过期消息。Compact 的过程就是扫描日志的所有消息，剔除那些过期的消息，然后把剩下的消息整理在一起。
  

• 我在这里贴一张来自官网的图片，来说明 Compact 过程。图中位移为 0、2 和 3 的消息的 Key 都是 K1。Compact 之后，分区只需要保存位移为 3 的消息，因为它是最新发送的。

• Kafka 提供了专门的后台线程定期地巡检待 Compact 的主题，看看是否存在满足条件的可删除数据。这个后台线程叫 Log Cleaner。很多实际生产环境中都出现过位移主题无限膨胀占用过多磁盘空间的问题，如果你的环境中也有这个问题，我建议你去检查一下 Log Cleaner 线程的状态，通常都是这个线程挂掉了导致的。

九、重平衡

1. 背景

重平衡指相同group下的所有consumer如何分配某topic下的所有分区达成共识的过程。在重平衡的过程中，所有消费者实例参加，需要借助协调组件，分配某主题下的所有分区，此过程是无法消费消息的，导致消息大量堆积，影响实时性。

• 在kafka中，将协调者称为Coordinator，它专门为 Consumer Group 服务，负责为 Group 执行 Rebalance 以及提供位移管理和组成员管理等。
• 具体来讲，consumer向broker提交位移时，实际上是向group属于的Coordinator所在brokcer提交位移。consumer启动时，向Coordinator发送请求，Coordinator负责分配分区给consumer。

2. Kafka 为某个 Consumer Group 确定 Coordinator 所在的 Broker 的算法有 2 个步骤。

• 第 1 步：确定由位移主题的哪个分区来保存该 Group 数据：partitionId=Math.abs(groupId.hashCode() % offsetsTopicPartitionCount)。
• 第 2 步：找出该分区 Leader 副本所在的 Broker，该 Broker 即为对应的 Coordinator。
• 举例。首先，Kafka 会计算该 Group 的 group.id 参数的哈希值。比如你有个 Group 的 group.id 设置成了“test-group”，那么它的 hashCode 值就应该是 627841412。其次，Kafka 会计算 __consumer_offsets 的分区数，通常是 50 个分区，之后将刚才那个哈希值对分区数进行取模加求绝对值计算，即 abs(627841412 % 50) = 12。此时，我们就知道了位移主题的分区 12 负责保存这个 Group 的数据。有了分区号，算法的第 2 步就变得很简单了，我们只需要找出位移主题分区 12 的 Leader 副本在哪个 Broker 上就可以了。这个 Broker，就是我们要找的 Coordinator。

3. 重平衡发生的条件

• 消费者实例变化
• 订阅的topic分区变化
• 订阅的topic数量变化
• Consumer 端的 GC

4. 如何避免

• session.timeout.ms，就是被用来表征此事的。该参数的默认值是 10 秒，即如果 Coordinator 在 10 秒之内没有收到 Group 下某 Consumer 实例的心跳，它就会认为这个 Consumer 实例已经挂了。可以这么说，session.timeout.ms 决定了 Consumer 存活性的时间间隔。
• 控制发送心跳请求频率的参数，就是 heartbeat.interval.ms。这个值设置得越小，Consumer 实例发送心跳请求的频率就越高。
• 控制 Consumer 实际消费能力对 Rebalance 的影响，即 max.poll.interval.ms 参数。它限定了 Consumer 端应用程序两次调用 poll 方法的最大时间间隔。它的默认值是 5 分钟，表示你的 Consumer 程序如果在 5 分钟之内无法消费完 poll 方法返回的消息，那么 Consumer 会主动发起“离开组”的请求，Coordinator 也会开启新一轮 Rebalance。
• 合理设置GC参数

十、分批提交位移

• 消费者位移切记是下一条消息的位移，而不是目前最新消费消息的位移。

之前介绍过位移的提交。主要分为自动提交和手动提交（同步、异步）.

• 在手动提交位移时，会面临一个问题。当批量消费消息时，是全部处理完提交位移，还是处理一批提交一次。举个例子，一次性消费1000条数据，如果处理完1000条提交位移，中间过程除了差错，将不会提交位移，有大量的消息被重复消费。所以建议是分批提交位移。
• 以 commitAsync 为例，展示一段代码，实际上，commitSync 的调用方法和它是一模一样的。

private Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
int count = 0;
……
while (true) 
            ConsumerRecords<String, String> records = 
  consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record: records) 
                        process(record);  // 处理消息
                        offsets.put(new TopicPartition(record.topic(), record.partition()),
                                   new OffsetAndMetadata(record.offset() + 1)；
                       if（count % 100 == 0）
                                    consumer.commitAsync(offsets, null); // 回调处理逻辑是null
                        count++;
  

简单解释一下这段代码。程序先是创建了一个 Map 对象，用于保存 Consumer 消费处理过程中要提交的分区位移，之后开始逐条处理消息，并构造要提交的位移值。还记得之前我说过要提交下一条消息的位移吗？这就是这里构造 OffsetAndMetadata 对象时，使用当前消息位移加 1 的原因。代码的最后部分是做位移的提交。我在这里设置了一个计数器，每累计 100 条消息就统一提交一次位移。与调用无参的 commitAsync 不同，这里调用了带 Map 对象参数的 commitAsync 进行细粒度的位移提交。这样，这段代码就能够实现每处理 100 条消息就提交一次位移，不用再受 poll 方法返回的消息总数的限制了。

十一、CommitFailedException

CommitFailedException异常通常发生在手动提交位移中，大概意思就是手动同步提交位移失败，手动异步是不能捕获到异常的
session.timeout.ms是接收心跳包的最大时长
heartbeat.interval.ms是发送心跳包的时长
max.poll.interval.ms是两次poll的最大间隔时间

1. 当两次调用poll的间隔大于max.poll.interval.ms设置的时长，broker会认为这个消费者实例下线，会重平衡。但是这个消费者还会poll消息，就会导致异常。官方给出了两种解决办法：

• 延长max.poll.interval.ms时间
• 缩短批量处理消息的数量：max.poll.records

2. 其实最有效的处理方式是将处理消息作为异步，用多线程去执行。但是消费者不是线程安全的，在多线程处理时可能会发生并发的问题
3. 还有一种情况，也会产生CommitFailedException。kafka是支持独立消费者的，也就是Standalone Consumer。consumer.assign的消费方式，就是代表这个消费者是独立消费者，独立消费者也要指定 group.id 参数才能提交位移。

• 现在问题来了，如果你的应用中同时出现了设置相同 group.id 值的消费者组程序和独立消费者程序，那么当独立消费者程序手动提交位移时，Kafka 就会立即抛出 CommitFailedException 异常，因为 Kafka 无法识别这个具有相同 group.id 的消费者实例，于是就向它返回一个错误，表明它不是消费者组内合法的成员。

十二、多线程消费者开发实例

KafkaConsumer类不是线程安全的（thread-safe）。所有的网络I/O处理都是发生在用户主线程中，所以不能在多线程中共享同一个KafkaConsumer实例，否则程序会抛ConcurrentModificationException异常。当然有办法去实现多线程，目前有两种方案

1. 启用多个kafka consumer，每个consumer单独消费和处理消息。

• 这样的话，就相当于多个消费者去分摊多个分区的消息。举个例子，现在有100个分区，启用10个消费者，每个消费者会分配到10个分区。但是会有约束性，上面这种情况，consumer有效值最大为100个，并且一旦consumer发生异常，重平衡将会非常花费时间。

public class KafkaConsumerRunner implements Runnable 
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final KafkaConsumer consumer;


     public void run() 
         try 
             consumer.subscribe(Arrays.asList("topic"));
             while (!closed.get()) 
      ConsumerRecords records = 
        consumer.poll(Duration.ofMillis(10000));
                 //  执行消息处理逻辑
             
          catch (WakeupException e) 
             // Ignore exception if closing
             if (!closed.get()) throw e;
          finally 
             consumer.close();
         
     


     // Shutdown hook which can be called from a separate thread
     public void shutdown() 
         closed.set(true);
         consumer.wakeup();
     

2. 启用单个消费者，使用多线程对这个消费者的消息进行处理。也就是单线程消费，多线程处理，将消费和处理解耦。

• 这种方式消息消费和处理顺序将会打乱，并且需要维护线程池。

private final KafkaConsumer<String, String> consumer;
private ExecutorService executors;
...


private int workerNum = ...;
executors = new ThreadPoolExecutor(
  workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
  new ArrayBlockingQueue<>(1000), 
  new ThreadPoolExecutor.CallerRunsPolicy());


...
while (true)  
  ConsumerRecords<String, String> records = 
    consumer.poll(Duration.ofSeconds(1));
  for (final ConsumerRecord record : records) 
    executors.submit(new Worker(record));
  

..

十二、Consumer管理tcp连接

1. 发起 FindCoordinator 请求时。还记得消费者端有个组件叫协调者（Coordinator）吗？它驻留在 Broker 端的内存中，负责消费者组的组成员管理和各个消费者的位移提交管理。当消费者程序首次启动调用 poll 方法时，它需要向 Kafka 集群发送一个名为 FindCoordinator 的请求，希望 Kafka 集群告诉它哪个 Broker 是管理它的协调者。不过，消费者应该向哪个 Broker 发送这类请求呢？理论上任何一个 Broker 都能回答这个问题，也就是说消费者可以发送 FindCoordinator 请求给集群中的任意服务器。在这个问题上，社区做了一点点优化：消费者程序会向集群中当前负载最小的那台 Broker 发送请求。负载是如何评估的呢？其实很简单，就是看消费者连接的所有 Broker 中，谁的待发送请求最少。当然了，这种评估显然是消费者端的单向评估，并非是站在全局角度，因此有的时候也不一定是最优解。不过这不并影响我们的讨论。总之，在这一步，消费者会创建一个 Socket 连接。
2. 连接协调者时。Broker 处理完上一步发送的 FindCoordinator 请求之后，会返还对应的响应结果（Response），显式地告诉消费者哪个 Broker 是真正的协调者，因此在这一步，消费者知晓了真正的协调者后，会创建连向该 Broker 的 Socket 连接。只有成功连入协调者，协调者才能开启正常的组协调操作，比如加入组、等待组分配方案、心跳请求处理、位移获取、位移提交等。
3. 消费数据时。消费者会为每个要消费的分区创建与该分区领导者副本所在 Broker 连接的 TCP。举个例子，假设消费者要消费 5 个分区的数据，这 5 个分区各自的领导者副本分布在 4 台 Broker 上，那么该消费者在消费时会创建与这 4 台 Broker 的 Socket 连接。

十三、kafka副本机制详解

kafka副本是建立在分区之上的

1. 副本的作用

• 提供数据冗余，假设broker1宕机，broker2作为副本可以立即提供服务
• 读写分离，一般来说主节点负责写，从节点负责读，但是kafka的读写全部在leader副本
• 为用户选择邻近副本，提高响应速度，kafka也没做到这一步

2. kafka的follower副本不提供读写的原因：

• 当你使用生产者 API 向 Kafka 成功写入消息后，马上使用消费者 API 去读取刚才生产的消息，如果是从follower读取，follower同步leader的消息会不及时
• 如果支持从follower读，那么可能follower1和follower2同步leader消息的情况不一致，造成一会儿看得到消息，一会儿又看不到了

3. follower从leader异步拉取消息，会造成消息同步不及时的情况，怎么判断消息同步是否及时，这里会涉及到一个ISR集合。

• In-sync Replicas，也就是所谓的 ISR 副本集合。ISR 中的副本都是与 Leader 同步的副本，相反，不在 ISR 中的追随者副本就被认为是与 Leader 不同步的。那么，到底什么副本能够进入到 ISR 中呢？我们首先要明确的是，Leader 副本天然就在 ISR 中。也就是说，ISR 不只是追随者副本集合，它必然包括 Leader 副本。甚至在某些情况下，ISR 只有 Leader 这一个副本。
• Kafka 判断 Follower 是否与 Leader 同步的标准，不是看相差的消息数，而是根据Broker 端参数 replica.lag.time.max.ms 参数值。这个参数的含义是 Follower 副本能够落后 Leader 副本的最长时间间隔，当前默认值是 10 秒。这就是说，只要一个 Follower 副本落后 Leader 副本的时间不连续超过 10 秒，那么 Kafka 就认为该 Follower 副本与 Leader 是同步的，即使此时 Follower 副本中保存的消息明显少于 Leader 副本中的消息。
• 倘若该副本后面慢慢地追上了 Leader 的进度，那么它是能够重新被加回 ISR 的。这也表明，ISR 是一个动态调整的集合，而非静态不变的。

4. Unclean 领导者选举（Unclean Leader Election）

• 既然 ISR 是可以动态调整的，那么自然就可以出现这样的情形：ISR 为空。因为 Leader 副本天然就在 ISR 中，如果 ISR 为空了，就说明 Leader 副本也“挂掉”了，Kafka 需要重新选举一个新的 Leader。可是 ISR 是空，此时该怎么选举新 Leader 呢？
• Kafka 把所有不在 ISR 中的存活副本都称为非同步副本。通常来说，非同步副本落后 Leader 太多，因此，如果选择这些副本作为新 Leader，就可能出现数据的丢失。毕竟，这些副本中保存的消息远远落后于老 Leader 中的消息。在 Kafka 中，选举这种副本的过程称为 Unclean 领导者选举。Broker 端参数 unclean.leader.election.enable 控制是否允许 Unclean 领导者选举。
• 开启 Unclean 领导者选举可能会造成数据丢失，但好处是，它使得分区 Leader 副本一直存在，不至于停止对外提供服务，因此提升了高可用性。反之，禁止 Unclean 领导者选举的好处在于维护了数据的一致性，避免了消息丢失，但牺牲了高可用性。