kafka的ISR机制被成为“不丢消息”机制。在说ISR机制前，先讲一下kafka的副本（replica）。

kafka的Replica

1.kafka的topic可以设置有N个副本（replica），副本数最好要小于broker的数量，也就是要保证一个broker上的replica最多有一个，所以可以用broker id指定Partition replica。

2.创建副本的单位是topic的分区，每个分区有1个leader和0到多个follower，我们把多个replica分为Lerder replica和follower replica。

3.当producer在向partition中写数据时，根据ack机制，默认ack=1，只会向leader中写入数据，然后leader中的数据会复制到其他的replica中，follower会周期性的从leader中pull数据，但是对于数据的读写操作都在leader replica中，follower副本只是当leader副本挂了后才重新选取leader，follower并不向外提供服务。

kafka的“同步”

kafka不是完全同步，也不是完全异步，是一种特殊的ISR（In Sync Replica）

1.leader会维持一个与其保持同步的replica集合，该集合就是ISR，每一个partition都有一个ISR，它时有leader动态维护。

2.我们要保证kafka不丢失message，就要保证ISR这组集合存活（至少有一个存活），并且消息commit成功。

所以我们判定存活的概念时什么呢？分布式消息系统对一个节点是否存活有这样两个条件判断：第一个，节点必须维护和zookeeper的连接，zookeeper通过心跳机制检查每个节点的连接；第二个，如果节点时follower，它必要能及时同步与leader的写操作，不是延时太久。

如果满足上面2个条件，就可以说节点时“in-sync“（同步中的）。leader会追踪”同步中的“节点，如果有节点挂了，卡了，或延时太久，那么leader会它移除，延时的时间由参数replica.log.max.messages决定，判断是不是卡住了，由参数replica.log.time.max.ms决定。

kafka的commit是由ack机制决定的，下一节再讲ack机制。

一、kafka replica

1. 当某个topic的replication-factor为N且N大于1时，每个Partition都会有N个副本(Replica)。kafka的replica包含leader与follower。

2. Replica的个数小于等于Broker的个数，也就是说，对于每个Partition而言，每个Broker上最多只会有一个Replica，因此可以使用Broker id 指定Partition的Replica。

3. 所有Partition的Replica默认情况会均匀分布到所有Broker上。

二、Data Replication如何Propagate(扩散出去)消息？

每个Partition有一个leader与多个follower，producer往某个Partition中写入数据是，只会往leader中写入数据，然后数据才会被复制进其他的Replica中。
数据是由leader push过去还是有flower pull过来？
kafka是由follower周期性或者尝试去pull(拉)过来(其实这个过程与consumer消费过程非常相似)，写是都往leader上写，但是读并不是任意flower上读都行，读也只在leader上读，flower只是数据的一个备份，保证leader被挂掉后顶上来，并不往外提供服务。

三、Data Replication何时Commit？

同步复制： 只有所有的follower把数据拿过去后才commit，一致性好，可用性不高。
异步复制： 只要leader拿到数据立即commit，等follower慢慢去复制，可用性高，立即返回，一致性差一些。
Commit：是指leader告诉客户端，这条数据写成功了。kafka尽量保证commit后立即leader挂掉，其他flower都有该条数据。

kafka不是完全同步，也不是完全异步，是一种ISR机制：
1. leader会维护一个与其基本保持同步的Replica列表，该列表称为ISR(in-sync Replica)，每个Partition都会有一个ISR，而且是由leader动态维护
2. 如果一个flower比一个leader落后太多，或者超过一定时间未发起数据复制请求，则leader将其重ISR中移除
3. 当ISR中所有Replica都向Leader发送ACK时，leader才commit

既然所有Replica都向Leader发送ACK时，leader才commit，那么flower怎么会leader落后太多？
producer往kafka中发送数据，不仅可以一次发送一条数据，还可以发送message的数组；批量发送，同步的时候批量发送，异步的时候本身就是就是批量；底层会有队列缓存起来，批量发送，对应broker而言，就会收到很多数据(假设1000)，这时候leader发现自己有1000条数据，flower只有500条数据，落后了500条数据，就把它从ISR中移除出去，这时候发现其他的flower与他的差距都很小，就等待；如果因为内存等原因，差距很大，就把它从ISR中移除出去。

commit策略：
server配置

  rerplica.lag.time.max.ms=10000
  # 如果leader发现flower超过10秒没有向它发起fech请求，那么leader考虑这个flower是不是程序出了点问题
  # 或者资源紧张调度不过来，它太慢了，不希望它拖慢后面的进度，就把它从ISR中移除。

  rerplica.lag.max.messages=4000 # 相差4000条就移除
  # flower慢的时候，保证高可用性，同时满足这两个条件后又加入ISR中，
  # 在可用性与一致性做了动态平衡   亮点1234567

topic配置

  min.insync.replicas=1 # 需要保证ISR中至少有多少个replica1

Producer配置

  request.required.asks=0
  # 0:相当于异步的，不需要leader给予回复，producer立即返回，发送就是成功,
      那么发送消息网络超时或broker crash(1.Partition的Leader还没有commit消息 2.Leader与Follower数据不同步)，
      既有可能丢失也可能会重发  # 1：当leader接收到消息之后发送ack，丢会重发，丢的概率很小
  # -1：当所有的follower都同步消息成功后发送ack.  丢失消息可能性比较低123456

四、Data Replication如何处理Replica恢复

leader挂掉了，从它的follower中选举一个作为leader，并把挂掉的leader从ISR中移除，继续处理数据。一段时间后该leader重新启动了，它知道它之前的数据到哪里了，尝试获取它挂掉后leader处理的数据，获取完成后它就加入了ISR。

五、Data Replication如何处理Replica全部宕机

1、等待ISR中任一Replica恢复,并选它为Leader

1. 等待时间较长,降低可用性

2. 或ISR中的所有Replica都无法恢复或者数据丢失,则该Partition将永不可用

2、选择第一个恢复的Replica为新的Leader,无论它是否在ISR中

1. 并未包含所有已被之前Leader Commit过的消息,因此会造成数据丢失

2. 可用性较高

Kafka 的 Data Replication 需要解决如下问题：

怎样 Propagate 消息

在向 Producer 发送 ACK 前需要保证有多少个 Replica 已经收到该消息

怎样处理某个 Replica 不工作的情况

怎样处理 Failed Replica 恢复回来的情况

2. Propagate 消息

通过zookeeper先知道leader在哪一台机器上，然后produce将消息发送到leader上，Follower 在收到该消息并写入其 Log 后，向 Leader 发送 ACK。一旦 Leader 收到了 ISR 中的所有 Replica 的 ACK，该消息就被认为已经 commit 了，Leader 将增加 HW 并且向 Producer 发送 ACK。

3. ACK 前需要保证有多少个 Replica 已经收到该消息

Leader 会跟踪与其保持同步的 Replica 列表，该列表称为 ISR(即 in-sync Replica)。如果一个 Follower 宕机，或者落后太多，Leader 将把它从 ISR 中移除。

Kafka 的复制机制既不是完全的同步复制，也不是单纯的异步复制。完全同步复制要求所有能工作的 Follower 都复制完，这条消息才会被认为 commit，这种复制方式极大的影响了吞吐率(高吞吐率是 Kafka 非常重要的一个特性)。而异步复制方式下，Follower 异步的从 Leader 复制数据，数据只要被 Leader 写入 log 就被认为已经 commit，这种情况下如果 Follower 都复制完都落后于 Leader，而如果 Leader 突然宕机，则会丢失数据。

4. Data Replication如何处理Replica全部宕机

1、等待ISR中任一Replica恢复,并选它为Leader

等待时间较长,降低可用性

或ISR中的所有Replica都无法恢复或者数据丢失,则该Partition将永不可用

2、选择第一个恢复的Replica为新的Leader,无论它是否在ISR中

并未包含所有已被之前Leader Commit过的消息,因此会造成数据丢失

可用性较高

5. Data Replication如何处理Replica恢复

leader挂掉了，从它的follower中选举一个作为leader，并把挂掉的leader从ISR中移除，继续处理数据。一段时间后该leader重新启动了，它知道它之前的数据到哪里了，尝试获取它挂掉后leader处理的数据，获取完成后它就加入了ISR。

6. ack机制

方案 优点 缺点

半数以上完成同步，就发送ack 延迟低 选举新的leader时，容忍n台节点的故障，需要2n+1个副本

全部完成同步，才发送ack 选举新的leader时，容忍n台节点的故障，需要n+1个副本 延迟高

7. Exactly Once

在0.11版本之后，Kafka引入了幂等性机制(idempotent)，配合acks = -1时的at least once语义，实现了producer到broker的exactly once语义。

idempotent + at least once = exactly once

使用时，只需将enable.idempotence属性设置为true，kafka自动将acks属性设为-1。

ISR（in-sync replica） 就是 Kafka 为某个分区维护的一组同步集合，即每个分区都有自己的一个 ISR 集合，处于 ISR 集合中的副本，意味着 follower 副本与 leader 副本保持同步状态，只有处于 ISR 集合中的副本才有资格被选举为 leader。一条 Kafka 消息，只有被 ISR 中的副本都接收到，才被视为“已同步”状态。这跟 zk 的同步机制不一样，zk 只需要超过半数节点写入，就可被视为已写入成功。

follwer 副本与 leader 副本之间的数据同步流程如下：

这也就意味着，leader 副本永远领先 follower 副本，且各个 follower 副本之间的消息最新位移也不尽相同，Kafka 必须要定义一个落后 leader 副本位移的范围，使得处于这个范围之内的 follower 副本被认为与 leader 副本是处于同步状态的，即处于 ISR 集合中。

ISR (In-Sync Replicas)是Leader在Zookeeper中动态维护基本保持同步的Replica列表，该列表中保存的是与Leader副本保持消息同步的所有副本对应的Follower节点id。ISR冗余备份机制核心逻辑围绕HW值、LEO值展开。

1、broker 收到producer的请求

2、leader 收到消息，并成功写入，LEO 值+1

3、broker 将消息推给follower replica，follower 成功写入 LEO +1

…

4、所有LEO 写入后，leader HW +1

5、消息可被消费，并成功响应

在Kafka上创建一个Topic的步骤：进入服务器后，找到kafka安装目录进入bin文件夹，输入命令--- 查看kafka现有主题命令:。/kafka-topics.sh --list --zookeeper zk_host:port望采纳。

所以我们判定存活的概念时什么呢？分布式消息系统对一个节点是否存活有这样两个条件判断：第一个，节点必须维护和zookeeper的连接，zookeeper通过心跳机制检查每个节点的连接；第二个，如果节点时follower，它必要能及时同步与leader的写操作，不是延时太久。

如果满足上面2个条件，就可以说节点时“in-sync“（同步中的）。leader会追踪”同步中的“节点，如果有节点挂了，卡了，或延时太久，那么leader会它移除，延时的时间由参数replica.log.max.messages决定，判断是不是卡住了，由参数replica.log.time.max.ms决定。

kafka的commit是由ack机制决定的，下一节再讲ack机制