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
image.pngLogEndOffset 的缩写,表示每个 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();
}
image.png
总结
- 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该消息。
网友评论