2018-12-03
Kafka消息保证生产的信息不丢失和重复消费问题
1）使用同步模式的时候，有3种状态保证消息被安全生产，在配置为1（只保证写入leader成功）的话，如果刚好leader partition挂了，数据就会丢失。
2）还有一种情况可能会丢失消息，就是使用异步模式的时候，当缓冲区满了，如果配置为0（还没有收到确认的情况下，缓冲池一满，就清空缓冲池里的消息），
数据就会被立即丢弃掉。
在数据生产时避免数据丢失的方法：
只要能避免上述两种情况，那么就可以保证消息不会被丢失。
1）就是说在同步模式的时候，确认机制设置为-1，也就是让消息写入leader和所有的副本。
2）还有，在异步模式下，如果消息发出去了，但还没有收到确认的时候，缓冲池满了，在配置文件中设置成不限制阻塞超时的时间，也就说让生产端一直阻塞，这样也能保证数据不会丢失。
在数据消费时，避免数据丢失的方法：如果使用了storm，要开启storm的ackfail机制；如果没有使用storm，确认数据被完成处理之后，再更新offset值。低级API中需要手动控制offset值。
推荐配置
producer
block.on.buffer.full = true 尽管该参数在0.9.0.0已经被标记为“deprecated”，但鉴于它的含义非常直观，所以这里还是显式设置它为true，使得producer将一直等待缓冲区直至其变为可用。否则如果producer生产速度过快耗尽了缓冲区，producer将抛出异常。缓冲区满了就阻塞在那，不要抛异常，也不要丢失数据
acks=all 很好理解，所有follower都响应了才认为消息提交成功，即"committed"
retries = MAX 无限重试，直到你意识到出现了问题
max.in.flight.requests.per.connection = 1 限制客户端在单个连接上能够发送的未响应请求的个数。设置此值是1表示kafka broker在响应请求之前client不能再向同一个broker发送请求。注意：设置此参数是为了避免消息乱序
使用KafkaProducer.send(record, callback)而不是send(record)方法 自定义回调逻辑处理消息发送失败，比如记录在日志中，用定时脚本扫描重处理
callback逻辑中最好显式关闭producer：close(0) 注意：设置此参数是为了避免消息乱序（仅仅因为一条消息发送没收到反馈就关闭生产者，感觉代价很大）
unclean.leader.election.enable=false 关闭unclean leader选举，即不允许非ISR中的副本被选举为leader，以避免数据丢失
replication.factor >= 3 这个完全是个人建议了，参考了Hadoop及业界通用的三备份原则
min.insync.replicas > 1 消息至少要被写入到这么多副本才算成功，也是提升数据持久性的一个参数。与acks配合使用
保证replication.factor > min.insync.replicas 如果两者相等，当一个副本挂掉了分区也就没法正常工作了。通常设置replication.factor = min.insync.replicas + 1即可
consumer
enable.auto.commit=false 关闭自动提交位移
在消息被完整处理之后再手动提交位移
https://blog.csdn.net/weixin_38750084/article/details/82939435
kafka动态维护了一个同步状态的副本的集合（a set of In-Sync Replicas），简称ISR
ack
ack确认机制设置为0，表示不等待响应，不等待borker的确认信息，最小延迟，producer无法知道消息是否发生成功，消息可能丢失，但具有最大吞吐量。
ack确认机制设置为-1，也就是让消息写入leader和所有的副本，ISR列表中的所有replica都返回确认消息。
ack确认机制设置为1，leader已经接收了数据的确认信息，replica异步拉取消息，比较折中。
ack确认机制设置为2，表示producer写partition leader和其他一个follower成功的时候，broker就返回成功，无论其他的partition follower是否写成功。
ack确认机制设置为 "all" 即所有副本都同步到数据时send方法才返回, 以此来完全判断数据是否发送成功, 理论上来讲数据不会丢失。
min.insync.replicas=1 意思是至少有1个replica返回成功，否则product异常
topic
LogEndOffset 的缩写，表示每个 partition 的 log 最后一条 Message 的位置。HW 是 HighWatermark 的缩写，是指 consumer 能够看到的此 partition 的位置
leader 负责维护和跟踪 ISR(In-Sync Replicas 的缩写，表示副本同步队列) 中所有 follower 滞后的状态。
ISR 是 AR 中的一个子集，由 leader 维护 ISR 列表，follower 从 leader 同步数据有一些延迟（包括延迟时间 replica.lag.time.max.ms 和延迟条数 replica.lag.max.messages 两个维度, 当前最新的版本 0.10.x 中只支持 replica.lag.time.max.ms 这个维度），任意一个超过阈值都会把 follower 剔除出 ISR, 存入 OSR（Outof-Sync Replicas）列表，新加入的 follower 也会先存放在 OSR 中。AR=ISR+OSR。
HW 俗称高水位，HighWatermark 的缩写，取一个 partition 对应的 ISR 中最小的 LEO 作为 HW，consumer 最多只能消费到 HW 所在的位置。另外每个 replica 都有 HW,leader 和 follower 各自负责更新自己的 HW 的状态。对于 leader 新写入的消息，consumer 不能立刻消费，leader 会等待该消息被所有 ISR 中的 replicas 同步后更新 HW，此时消息才能被 consumer 消费。这样就保证了如果 leader 所在的 broker 失效，该消息仍然可以从新选举的 leader 中获取。对于来自内部 broKer 的读取请求，没有 HW 的限制。
有两个地方会对kafka的 Zookeeper 节点进行维护：
Controller 来维护：Kafka 集群中的其中一个 Broker 会被选举为 Controller，主要负责 Partition 管理和副本状态管理，也会执行类似于重分配 partition 之类的管理任务。在符合某些特定条件下，Controller 下的 LeaderSelector 会选举新的 leader，ISR 和新的 leader_epoch 及 controller_epoch 写入 Zookeeper 的相关节点中。同时发起 LeaderAndIsrRequest 通知所有的 replicas。
leader 来维护：leader 有单独的线程定期检测 ISR 中 follower 是否脱离 ISR, 如果发现 ISR 变化，则会将新的 ISR 的信息返回到 Zookeeper 的相关节点中。
如果某一个 partition 的所有 replica 都挂了，就无法保证数据不丢失了。这种情况下有两种可行的方案：
等待 ISR 中任意一个 replica“活”过来，并且选它作为 leader
选择第一个“活”过来的 replica（并不一定是在 ISR 中）作为 leader (默认情况)
https://www.infoq.cn/article/depth-interpretation-of-kafka-data-reliability
LSO：是 LastStableOffset 的简称，对未完成的事务而言，LSO 的值等于事务中第一条消息的位置(firstUnstableOffset)，对已完成的事务而言，它的值同 HW 相同
LW：Low Watermark 低水位, 代表 AR 集合中最小的 logStartOffset 值
如何实现高吞吐量
- 顺序读写
- 零拷贝
- 文件分段
- 批量发送
- 数据压缩
幂等性
PID。每个新的Producer在初始化的时候会被分配一个唯一的PID，这个PID对用户是不可见的。
Sequence Numbler。（对于每个PID，该Producer发送数据的每个<Topic, Partition>都对应一个从0开始单调递增的Sequence Number
enable.idempotence=true
只能保证单个Producer对于同一个<Topic, Partition>的Exactly Once语义。不能保证同一个Producer一个topic不同的partion幂等。
代码
- 调用kafkaProducer的send方法将数据添加到 RecordAccumulator 中，添加时会判断是否需要新建一个 ProducerBatch，这时这个 ProducerBatch 还是没有 PID 和 sequence number 信息的；
- Producer 后台发送线程 Sender，在 run() 方法中，会先根据 TransactionManager 的 shouldResetProducerStateAfterResolvingSequences() 方法判断当前的 PID 是否需要重置，重置的原因是因为：如果有topic-partition的batch已经超时还没处理完，此时可能会造成sequence number 不连续。因为sequence number 有部分已经分配出去了，而Kafka服务端没有收到这部分sequence number 的序号，Kafka服务端为了保证幂等性，只会接受同一个pid的sequence number 等于服务端缓存sequence number +1的消息，所有这时候需要重置Pid来保证幂等性
- Sender线程调用maybeWaitForProducerId()方法判断是否要申请Pid，如果需要，会阻塞直到成功申请到Pid
- 最后调用sendProduceRequest方法将消息发送出去
相关类
BatchMetadata
用来存储Batch的元数据， BatchMetadata类的几个重要的字段
ProducerStateEntry
用于存储每个producerId对应的Batch，按照sequence从小到大进行排序，最小的作为头，最大的作为尾 ，每个producerId的队列失踪保持着最多5个Batch，如果超过5个了，就从头开始remove
ProducerAppendInfo用于在追加的消息写到Log之前进行校验，主要对epoch、sequence number进行校验
currentEntry：ProducerStateEntry类型，就是pid对应的ProducerStateEntry中batchMetadata尾部对象，用于跟新追加的Batch做比较
validationType：校验的方式。不同的类型，校验的规则不一样
checkSequence方法是一个跟幂等性很重要的方法，此方法就是校验sequence number的。有以下几个判断规则
- 如果producerEpoch更新了，则追加的Batch里的appendFirstSeq必须是0
- 当currentLastSeq为-1时，说明此生产者还没有成功追加过消息，appendFirstSeq也必须是0
- appendFirstSeq = currentLastSeq+1，或者当currentLastSeq达到Int的最大值Int.MaxValue时，appendFirstSeq为0
ProducerStateManager
用来管理Producer的状态，里面存储了各个生产者与ProducerStateEntry的对应关系。每个ProducerStateManager对应一个TopicPartition
partition assgin
如何分配partition到broker
- 副本因子不能大于 Broker 的个数；
- 第一个分区（编号为0）的第一个副本放置位置是随机从 brokerList 选择的；
- 其他分区的第一个副本放置位置相对于第0个分区依次往后移。也就是如果我们有5个 Broker，5个分区，假设第一个分区放在第四个 Broker 上，那么第二个分区将会放在第五个 Broker 上；第三个分区将会放在第一个 Broker 上；第四个分区将会放在第二个 Broker 上，依次类推；
- 剩余的副本相对于第一个副本放置位置其实是由 nextReplicaShift 决定的，而这个数也是随机产生的；
如何分配partition到consumer
Range策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。
使用RoundRobin策略有两个前提条件必须满足：
- 同一个Consumer Group里面的所有消费者的num.streams必须相等；
- 每个消费者订阅的主题必须相同。
file design
Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。
通过索引信息可以快速定位message和确定response的最大大小。
通过index元数据全部映射到memory，可以避免segment file的IO磁盘操作。
通过索引文件稀疏存储，可以大幅降低index文件元数据占用空间大小
事务性
有了Transaction ID后，Kafka可保证：
- 跨Session的数据幂等发送。当具有相同Transaction ID的新的Producer实例被创建且工作时，旧的且拥有相同Transaction ID的Producer将不再工作。
- 跨Session的事务恢复。如果某个应用实例宕机，新的实例可以保证任何未完成的旧的事务要么Commit要么Abort，使得新实例从一个正常状态开始工作
Kafka 0.11.0.0引入了一个服务器端的模块，名为Transaction Coordinator，用于管理Producer发送的消息的事务性。
该Transaction Coordinator维护Transaction Log，该log存于一个内部的Topic内。由于Topic数据具有持久性，因此事务的状态也具有持久性。
Producer并不直接读写Transaction Log，它与Transaction Coordinator通信，然后由Transaction Coordinator将该事务的状态插入相应的Transaction Log。
Transaction Log的设计与Offset Log用于保存Consumer的Offset类似。
为了区分写入Partition的消息被Commit还是Abort，Kafka引入了一种特殊类型的消息，即Control Message。该类消息的Value内不包含任何应用相关的数据，并且不会暴露给应用程序。它只用于Broker与Client间的内部通信。
对于Producer端事务，Kafka以Control Message的形式引入一系列的Transaction Marker。Consumer即可通过该标记判定对应的消息被Commit了还是Abort了，然后结合该Consumer配置的隔离级别决定是否应该将该消息返回给应用程序。
Producer<String, String> producer = new KafkaProducer<String, String>(props);
// 初始化事务，包括结束该Transaction ID对应的未完成的事务（如果有）
// 保证新的事务在一个正确的状态下启动
producer.initTransactions();
// 开始事务
producer.beginTransaction();
// 消费数据
ConsumerRecords<String, String> records = consumer.poll(100);
try{
// 发送数据
producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value"));
// 发送消费数据的Offset，将上述数据消费与数据发送纳入同一个Transaction内
producer.sendOffsetsToTransaction(offsets, "group1");
// 数据发送及Offset发送均成功的情况下，提交事务
producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
// 数据发送或者Offset发送出现异常时，终止事务
producer.abortTransaction();
} finally {
// 关闭Producer和Consumer
producer.close();
consumer.close();
}
总结
- PID与Sequence Number的引入实现了写操作的幂等性
- 写操作的幂等性结合At Least Once语义实现了单一Session内的Exactly Once语义
- Transaction Marker与PID提供了识别消息是否应该被读取的能力，从而实现了事务的隔离性
- Offset的更新标记了消息是否被读取，从而将对读操作的事务处理转换成了对写（Offset）操作的事务处理
- Kafka事务的本质是，将一组写操作（如果有）对应的消息与一组读操作（如果有）对应的Offset的更新进行同样的标记（即Transaction Marker）来实现事务中涉及的所有读写操作同时对外可见或同时对外不可见
- Kafka只提供对Kafka本身的读写操作的事务性，不提供包含外部系统的事务性
与zk比较
Zookeeper
Zookeeper的原子广播协议与两阶段提交以及Kafka事务机制有相似之处，但又有各自的特点
Kafka事务可COMMIT也可ABORT。而Zookeeper原子广播协议只有COMMIT没有ABORT。当然，Zookeeper不COMMIT某消息也即等效于ABORT该消息的更新。
Kafka存在多个Transaction Coordinator实例，扩展性较好。而Zookeeper写操作只能在Leader节点进行，所以其写性能远低于读性能。
Kafka事务是COMMIT还是ABORT完全取决于Producer即客户端。而Zookeeper原子广播协议中某条消息是否被COMMIT取决于是否有一大半FOLLOWER ACK该消息。
