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

顺序消费

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

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

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

但是消费者里可能会有多个线程来并发来处理消息。因为如果消费者是单线程消费数据，那么这个吞吐量太低了。而多个线程并发的话，顺序可能就乱掉了
写N个queue，将具有相同key的数据都存储在同一个queue，然后对于N个线程，每个线程分别消费一个queue即可。

丢数据

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

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

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

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

手动提交 offset 的方法有两种:分别是 commitSync(同步提交)和 commitAsync(异步提交)。两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交;不同点是， commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试(由不可控因素导致，也会出现提交失败);而 commitAsync 则没有失败重试机制，故有可能提交失败。

给 topic设置 replication.factor ，这个值必须大于 1，保证每个 partition 必须至少有 2 个副本
在 kafka 服务端设置 min.insync.replicas 参数，这个值必须大于 1，这个是要求一个leader至少感知到有至少一个follower还跟自己保持联系，没掉队，这样才能确保 leader挂了还有一个follower，保证至少一个 follower能和leader保持正常的数据同步。

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

Rebalance

RangeAssignor:

//计算每个consumer分到的partition数量
                int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
                //计算平均以后剩余partition数量
                int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();

                //从0开始作为Partition Index, 构造TopicPartition对象
                List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);
                for (int i = 0, n = consumersForTopic.size(); i < n; i++) {//对于当前这个topic的每一个consumer
                    //一定是前面几个consumer会被分配一个额外的TopicPartitiion
                    int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition);
                    int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);
                    assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length));
                }

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;

RoundRobin：

for(每一个TopicPartition)

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

StickyAssignor分配策略

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

分区的分配要尽可能的均匀；
分区的分配尽可能的与上次分配的保持相同。
当两者发生冲突时，第一个目标优先于第二个目标。鉴于这两个目标，StickyAssignor策略的具体实现要比RangeAssignor和RoundRobinAssignor这两种分配策略要复杂很多。

image-20200909225735741

image-20200909230725291

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

拿数据要写库，首先检查下主键，如果有数据，则不插入，进行一次update
如果是写 redis，就没问题，反正每次都是 set ，天然幂等性
生产者发送消息的时候带上一个全局唯一的id,消费者拿到消息后，先根据这个id去 redis里查一下，之前有没消费过，没有消费过就处理，并且写入这个 id 到 redis，如果消费过了，则不处理。
基于数据库的唯一键

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

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

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

ISR：

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 频繁收缩与扩张的问题了。

controller

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

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

主题管理（创建、删除、增加分区）
分区重分配
所有主题信息。包括具体的分区信息，比如领导者副本是谁，ISR 集合中有哪些副本等，所有 Broker 信息。包括当前都有哪些运行中的 Broker，哪些正在关闭中的 Broker 等。所有涉及运维任务的分区。包括当前正在进行 Preferred 领导者选举以及分区重分配的分区列表。

故障转移

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

img