Kafka的重复、丢数据及顺序消费等问题_服务器知识

①、kafka的顺序消息仅仅是通过partitionKey，将某类消息写入同一个partition，一个partition只能对应一个消费线程，以保证数据有序。

②、除了发送消息需要指定partitionKey外，producer和consumer实例化无区别。

③、kafka broker宕机，kafka会有自选择，所以宕机不会减少partition数量，也就不会影响partitionKey的sharding。

acks设置为0：broker接收消息立即返回，还没写入磁盘，容易丢失数据

acks设置为1：等待broker的ack，如果leader落盘了就返回ack，如果follower同步完成前leader挂了就会丢失未同步的数据（follower选举）

acks设置为-1：等待所有leader和follower都落盘后返回ack，如果follower已同步，但是broker返回ack前leader挂了，则会重复发送消息。

consumer自动提交offset，但其实未处理好消息，容易丢数据。可以选择手动提交，处理完后再提交offset

0.9版本的kafka改进了coordinator的设计，提出了group coordinator——每个consumer group都会被分配一个这样的coordinator用于组管理和位移管理。这个group coordinator比原来承担了更多的责任，比如组成员管理、位移提交保护机制等。当新版本consumer group的第一个consumer启动的时候，它会去和kafka server确定谁是它们组的coordinator。之后该group内的所有成员都会和该coordinator进行协调通信。显而易见，这种coordinator设计不再需要zookeeper了，性能上可以得到很大的提升。

每个 Group 都会选择一个 Coordinator 来完成自己组内各 Partition 的 Offset 信息，选择的规则如下： 1. 计算 Group 对应在 __consumer_offsets 上的 Partition 2. 根据对应的Partition寻找该Partition的leader所对应的Broker，该Broker上的Group Coordinator即就是该Group的Coordinator

numPartitionsPerConsumer=counsumer/partitions——》5/3=1，每个消费者至少被分配一个partition

consumersWithExtraPartition=counsumer%partitions——》5%3=2

i=0,start=0,length=2

i=1,start=2,length=2

i=2,start=4,length=1

如果是4个partitions和3个consumer

i=0,start=0,length=2

i=1,start=2,length=1

i=2,start=3,length=1

for(每一个TopicPartition)

以RoundRobin的方式选择一个订阅了这个Topic的Consumer，将这个TopicPartition分派给这个Consumer end

“sticky”这个单词可以翻译为“粘性的”，Kafka从0.11.x版本开始引入这种分配策略，它主要有两个目的：

为什么会重复消费：第一种可能是生产者重复发送消息。第二种可能是消费者手动提交时挂掉了，导致消费了数据但是没有提交offset。

为什么会丢失数据：第一种可能是ack非-1的情况下，follower未同步完全，leader挂了。第二种可能是消费者自动提交，但其实还没完成消费。

怎么保证生产者消息不重复，0.11后，生产者会生成pid，和一个sequence number，通过pid sequence number brokerid作为key，如果在partition中已经存在，则只持久化一条。且Producer重启可以通过TransactionID拿到原来的pid，所以可以跨会话的保持一致

保证顺序消费：需要保证顺序的消息发到同一个partition中，consumer会自己根据顺序消费

0.9.0.0 版本之前判断副本之间是否同步，主要是靠参数 replica.lag.max.messages 决定的，即允许 follower 副本落后 leader 副本的消息数量，超过这个数量后，follower 会被踢出 ISR。

replica.lag.max.messages 也很难在生产上给出一个合理值，如果给的小，会导致 follower 频繁被踢出 ISR，如果给的大，broker 发生宕机导致 leader 变更时，肯能会发生日志截断，导致消息严重丢失的问题。

在 0.9.0.0 版本之后，Kafka 给出了一个更好的解决方案，去除了 replica.lag.max.messages，，用 replica.lag.time.max.ms 参数来代替，该参数的意思指的是允许 follower 副本不同步消息的最大时间值，即只要在 replica.lag.time.max.ms 时间内 follower 有同步消息，即认为该 follower 处于 ISR 中，这就很好地避免了在某个瞬间生产者一下子发送大量消息到 leader 副本导致该分区 ISR 频繁收缩与扩张的问题了。

Kafka集群中多个broker，有一个会被选举为controller leader，负责管理整个集群中分区和副本的状态，比如partition的leader 副本故障，由controller 负责为该partition重新选举新的leader 副本；当检测到ISR列表发生变化，有controller通知集群中所有broker更新其MetadataCache信息；或者增加某个topic分区的时候也会由controller管理分区的重新分配工作

实际上，Broker 在启动时，会尝试去 ZooKeeper 中创建 /controller 节点。Kafka 当前选举控制器的规则是：第一个成功创建 /controller 节点的 Broker 会被指定为控制器。

故障转移

当 Broker 0 宕机后，ZooKeeper 通过 Watch 机制感知到并删除了 /controller 临时节点。之后，所有存活的 Broker 开始竞选新的控制器身份。Broker 3 最终赢得了选举，成功地在 ZooKeeper 上重建了 /controller 节点。之后，Broker 3 会从 ZooKeeper 中读取集群元数据信息，并初始化到自己的缓存中。

上一篇 <<<Kafka如何实现分区及指定分区消费

下一篇 >>>Kafka如何保证高吞吐量

Kafka的重复、丢数据及顺序消费等问题

发表评论

评论列表（0条）