一、Kafka概述
1.1 Kafka简介
Apache Kafka 是一个快速、可扩展的、高吞吐的、可容错的分布式“发布-订阅”消息系统，使用 Scala 与 Java 语言编写， 能够将消息从一个端点传递到另一个端点， 较之传统的消息中间件（例如 ActiveMQ、 RabbitMQ）， Kafka 具有高吞吐量、内置分区、支持消息副本和高容错的特性， 非常适合大规模消息处理应用程序。
1.2 特点
1.高吞吐量
Kafka 每秒可以生产约 25 万消息（50 MB），每秒处理 55 万消息（110 MB）
kafka 高吞吐率实现
Kafka 与其它 MQ 相比，其最大的特点就是高吞吐率。为了增加存储能力， Kafka 将所有的消息都写入到了低速大容的硬盘。按理说，这将导致性能损失，但实际上， kafka 仍可保持超高的吞吐率，性能并未受到影响。其主要采用了如下的方式实现了高吞吐率。
- 顺序读写： Kafka 将消息写入到了分区 partition 中，而分区中消息是顺序读写的。顺序
读写要远快于随机读写。 - 零拷贝：生产者、消费者对于 kafka 中消息的操作是采用零拷贝实现的。
- 批量发送： Kafka 允许使用批量消息发送模式。
- 消息压缩： Kafka 支持对消息集合进行压缩
2.持久化数据存储
可进行持久化操作。将消息持久化到磁盘，因此可用于批量消费，例如 ETL，以及实时应用程序。通过将数据持久化到硬盘以及 replication 防止数据丢失。
3.分布式系统易于扩展
所有的 producer、broker 和 consumer 都会有多个，均为分布式的。无需停机即可扩展机器。
4.客户端端状态维护
消息被处理的状态是在 consumer 端维护，而不是由 server 端维护。当失败时能自动平衡。
1.3 应用场景
1.用户的活动追踪
用户在网站的不同活动消息发布到不同的主题中心，然后可以对这些消息进行实时监测实时处理。当然，也可加载到 Hadoop 或离线处理数据仓库，对用户进行画像。像淘宝、京东这些大型的电商平台，用户的所有活动都是要进行追踪的。
2.日志聚合
kafka的特性决定它非常适合作为"日志收集中心"。application可以将操作日志"批量""异步"的发送到kafka集群中，而不是保存在本地或者DB中；kafka可以批量提交消息/压缩消息等，这对producer端而言，几乎感觉不到性能的开支。此时consumer端可以使hadoop等其他系统化的存储和分析系统。
3.限流削峰
4.消息系统
对于一些常规的消息系统，kafka是个不错的选择。partitons/replication和容错，可以使kafka具有良好的扩展性和性能优势。不过到目前为止，我们应该很清楚认识到，kafka并没有提供JMS中的"事务性""消息传输担保(消息确认机制)""消息分组"等企业级特性；kafka只能使用作为"常规"的消息系统，在一定程度上，尚未确保消息的发送与接收绝对可靠(比如，消息重发，消息发送丢失等)。
二、Kafka架构
2.1 基本术语
1.topic
主题。 在 Kafka 中，使用一个类别属性来划分消息的所属类，划分消息的这个类称为 topic。
topic 相当于消息的分类标签，是一个逻辑概念。
2.partition
分区。 每个 topic 包含一个或多个 partition，其是一个物理概念，对应到系统上就是一个或若干个目录。
partiiton 本身是一个 FIFO 队列，其中的消息是有序的。但，在 Partition 间无法保证消息的顺序性
3.segment
段。将 partition 进一步细分为了若干的 segment，每个 segment 文件的最大大小相等。
4.broker
Kafka 集群包含一个或多个服务器， 每个服务器节点称为一个 broker。
假设某个 topic 中有 N 个 partition，集群中有 M 个 Broker， broker 与 partition 间的数量
关系：
- 若 N>=M，且（N%M=0），则每个 broker 上会平均存储 N/M 个 partition。
- 若 N>M，且（N%M!=0），这其中会出现 broker 上分配的 partition 不平均的情况。这种
情况要避免。 - 若 N<M，这种情况会出现有的 broker 上没有分到 partition 的情况。
5.producer
生产者。 即消息的发布者，其会将某 topic 的消息发布到相应的 partition 中。
6.consumer
消费者。 可以从 broker 中读取消息。
一个消费者可以消费多个 topic 的消息。
一个消费者可以消费同一个 topic 中的多个 partition 消息。
一个 partition 允许多个无关消费者同时消费。
7.Consumer group
consumer group 是 kafka 提供的可扩展且具有容错性的消费者机制。组内可以有多个消费者，它们共享一个公共的 ID，即 group ID。 组内的所有消费者会协调在一起平均消费订阅主题的所有分区。
Kafka 可以保证在稳定状态下，每个 consumer 都属于一个 consumer group，一个 partition 中的消息只能被同一个 consumer group 中的一个 consumer 消费，而一个组内 consumer 只会消费某一个或几个特定的partition。当然，一个消息可以同时被多个 consumer group 消费。
即组间数据是共享的，组内数据是竞争的。
组中 consumer 数量与 partition 数量的对应关系如下。
这种设计方式的好处是：实现简单，弊端是：消息分配不均。
8.Rebalance
当消费者组中消费者数量发生变化，或 Topic 中的 partition 数量发生了变化时， partition的所有权会在消费者间转移，即 partition 会重新分配，这个过程称为再均衡 Rebalance。
再均衡能够给消费者组及 broker 集群带来高可用性和伸缩性，但在再均衡期间消费者是无法读取消息的， 即整个 broker 集群有一小段时间是不可用的。 因此要避免不必要的再均衡 。
9.Replicas of partition
分区副本。副本是一个分区的备份，是为了防止消息丢失而创建的分区的备份。保障 partition 的高可用。
10.Partition Leader
每个 partition 有多个副本，其中有且仅有一个作为 Leader， Leader 是当前负责消息读写的 partition。 即所有读写操作只能发生于 Leader 分区上， producer 和 consumer 只跟 leader 交互。
11.Partition Follower
所有 Follower 都需要从 Leader 同步消息， Follower 与 Leader 始终保持消息同步。
partition leader 与 follower 是主备关系，而非主从。
12.Broker Controller
Kafka 集群的多个 broker 中，有一个会被选举为 controller，负责管理整个集群中 partition和副本 replicas 的状态。
当 partition leader 宕机后，broker controller 会从 ISR 中选举出一个 Follower 做为新的leader。所谓选举就是从 ISR 中找到第一个 Follower，直接让其当选新的 leader。
Broker Controller 是由 zk 选举出来的，用来进行 leader election 以及 各种 failover。
13.Zookeeper
Zookeeper存储集群的 meta 信息，负责维护和协调 broker， 负责 Broker Controller 的选举。
14.Coordinator
Coordinator 一般指的是运行在每个 broker 上的 group Coordinator 进程，用于管理Consumer Group 中的各个成员，主要用于 offset 位移管理和 Rebalance。 一个 Coordinator 可以同时管理多个消费者组。
2.2 Topics、Producers、Consumers
1.Topics
一个topic是对一组消息的归纳。用来在kafka中为隔离不同的数据，从而实现一个kafka集群为多个需求服务。
对每个topic，Kafka 对它的日志进行了分区。
每个分区都由一系列有序的、不可变的消息组成，这些消息被连续的追加到分区中。
数据连续写入不能更改。
分区中的每个消息都有一个连续的序列号叫做offset，用来在分区中唯一的标识这个消息。
在一个可配置的时间段内，Kafka集群保留所有发布的消息，不管这些消息有没有被消费。比如，如果消息的保存策略被设置为2天，那么在一个消息被发布的两天时间内，它都是可以被消费的。之后它将被丢弃以释放空间。Kafka的性能是和数据量无关的常量级的，所以保留太多的数据并不是问题。
每个分区在Kafka集群的若干服务中都有副本，这样这些持有副本的服务可以共同处理数据和请求，副本数量是可以配置的。副本使Kafka具备了容错能力。
kafka中的分区是负载均衡和失败恢复的基本单位。
每个分区都由一个副本作为“leader”，零或若干副本作为“followers”，leader负责处理消息的读和写，followers则去复制leader。如果leader down了，followers中的一台则会自动成为leader。集群中的每个服务器都会同时扮演两个角色：作为它所持有的一部分分区的leader，同时作为其他分区的followers，这样集群就会据有较好的负载均衡。
标0的为Leader，其他为follower。服务器上，有不同分区的Leader和follower，分摊读写压力。
将日志分区可以达到以下目的：首先这使得每个日志的数量不会太大，可以在单个服务上保存。另外每个分区可以单独发布和消费，为并发操作topic提供了一种可能。
分区是负载均衡失败恢复分布式数据存储的基本单元。
2.Producers
Producer将消息发布到它指定的topic中，并负责决定发布到哪个分区。通常简单的由负载均衡机制随机选择分区，但也可以通过特定的分区函数选择分区。
3.Consumers
实际上每个consumer唯一需要维护的数据是消息在日志中的位置，也就是offset。这个offset由consumer来维护：一般情况下随着consumer不断的读取消息，这offset的值不断增加，但其实consumer可以以任意的顺序读取消息，比如它可以将offset设置成为一个旧的值来重读之前的消息。
以上特点的结合，使Kafka consumers非常的轻量级：它们可以在不对集群和其他consumer造成影响的情况下读取消息。你可以使用命令行来"tail"消息而不会对其他正在消费消息的consumer造成影响。
消费消息通常有两种模式：队列模式（queuing）和发布-订阅模式(publish-subscribe)。
-
队列模式：队列模式中，多个consumers可以同时从服务端读取消息，每个消息只被其中一个consumer读到；
-
发布订阅模式：发布-订阅模式中消息被广播到所有的consumer中。
Consumers可以加入一个consumer group，（组内竞争，组外共享）。组内的Consumer是一个竞争的关系，共同竞争一个topic内的消息，topic中的消息将被分发到组中的一个成员中，同一条消息只发往其中的一个消费者。同一组中的consumer可以在不同的程序中，也可以在不同的机器上。而如果有多个Consumer group来消费相同的Topic中的消息，则组和组之间是一个共享数据的状态，每一个组都可以获取到这个主题中的所有消息。
-
如果所有的consumer都在一个组中，这就成为了传统的队列模式，在各consumer中实现负载均衡。
-
如果所有的consumer都不在不同的组中，这就成为了发布-订阅模式，所有的消息都被分发到所有的consumer中。
-
更常见的是，每个topic都有若干数量的consumer组来消费，每个组都是一个逻辑上的“订阅者”，为了容错和更好的稳定性，每个组都由若干consumer组成，在组内竞争实现负载均衡。实现了组内竞争负载均衡，组间共享互不影响，这其实就是一个发布-订阅模式，只不过订阅者是个组而不是单个consumer。
三、 Kafka 工作原理与过程
3.1 消息路由策略
在通过 API 方式发布消息时，生产者是以 Record 为消息进行发布的。 Record 中包含 key与 value， value 才是我们真正的消息本身，而 key 用于路由消息所要存放的 Partition。消息要写入到哪个 Partition 并不是随机的，而是有路由策略的。
- 若指定了 partition，则直接写入到指定的 partition；
- 若未指定 partition 但指定了 key， 则通过对 key 的 hash 值与 partition 数量取模，该取模
结果就是要选出的 partition 索引； - 若 partition 和 key 都未指定， 则使用轮询算法选出一个 partition。
3.2 消息写入流程
消息生产者将消息发送给 broker，并形成最终的可供消费者消费的 log，是一个比较复杂的过程。
注意： producer 不是 zk 的客户端， 其是不能直接访问 zk 的。
-
producer 向 broker 集群提交连接请求，其所连接上的任意 broker 都会向其发送 broker controller 的通信 URL，即 broker controller 主机配置文件中的 listeners 地址。
broker 是 zk 的客户端可以直接访问 zk，从/controller 节点就可以读取到谁是当前 broker集群的 controller。
-
当 producer 指定了要生产消息的 topic 后，其会向 broker controller 发送请求，请求当前topic 中所有 partition 的 leader 列表地址。
-
broker controller在接收到请求后，会从zk中查找到指定topic的所有partition的leader，并返回给 producer
broker 可以从 zk 的/brokers/topics/hdu/0/status 节点中可以读取到该 0 号 partition 的 leader -
producer 在接收到 leader 列表地址后，根据消息路由策略找到当前要发送消息所要发送的 partition leader，然后将消息发送给该 leader
-
leader 将消息写入本地 log，并通知 ISR 中的 followers
-
ISR 中的 followers 从 leader 中同步消息后向 leader 发送 ACK
-
leader 收到所有 ISR 中的 followers 的 ACK 后，增加 HW，表示消费者已经可以消费到该位置了
-
当然， 若 leader 在等待的 followers 的 ACK 超时了， 发现还有 follower 没有发送 ACK，则会将该 follower 从 ISR 中清除， 然后增加 HW。
ISR指的是：比如有三个副本 ，编号是① ② ③ ，其中② 是Leader ① ③是Follower。假设在数据同步过程中，①跟上Leader,但是③出现故障或没有及时同步，则① ②是一个ISR，而③不是ISR成员。后期在Leader选举时，会用到ISR机制。会优先从ISR中选择Leader。
3.3 创建Topic流程
-
controller 在 ZooKeeper 的 /brokers/topics 节点上注册 watcher，当 topic 被创建，则 controller 会通过 watch 得到该 topic 的 partition/replica 分配。
-
controller从 /brokers/ids 读取当前所有可用的 broker 列表，对于 set_p 中的每一个 partition：
2.1 从分配给该 partition 的所有 replica（称为AR）中任选一个可用的 broker 作为新的 leader，并将AR设置为新的ISR
2.2 将新的 leader 和 ISR 写入 /brokers/topics/[topic]/partitions/[partition]/state
总结：Controller在创建一个新的Topic后，Controlle的作用就是从副本中选出Leader。
-
controller 通过 RPC 向相关的 broker 发送 LeaderAndISRRequest。
3.4 消息消费流程
生产者将消息发送到 topic 中，消费者即可对其进行消费，其消费过程如下：
-
consumer 向 broker 集群提交连接请求，其所连接上的任意 broker 都会向其发送 broker controller 的通信 URL， 即 broker controller 主机配置文件中的 listeners 地址
-
当 consumer 指定了要消费的 topic 后，其会向 broker controller 发送 poll 请求
-
broker controller 会为 consumer 分配一个或几个 partition leader，并将该 partitioin 的当前 offset 发送给 consumer
-
consumer 会按照 broker controller 分配的 partition 对其中的消息进行消费
-
当消费者消费完该条消息后，消费者会向 broker 发送一个该消息已被消费的反馈，即该消息的 offset
* 若为手动提交， 则可以是每消费完一条消息就提交一个 offset， 也可以是消费完这 一批消息后再提交一个 offset。 关键看你的代码怎么写。
* 若为自动提交， 则是提交最后一个消费消息的 offset。 -
当 broker 接到消费者的 offset 后， 会更新到相应的__consumer_offset 中
-
以上过程一直重复，直到消费者停止请求消息
-
消费者可以重置 offset，从而可以灵活消费存储在 broker 上的消息
四、Kafka的存储机制
kafka通过topic来分主题存放数据，主题内又有分区，分区还可以有多个副本 ，分区的内部还细分为若干个segment。
4.1 分区与备份
所谓的分区 其实就是在 kafka对应存储目录下创建的文件夹，文件夹的名字是主题名加上分区编号 编号从0开始。
-
建的 test 主题是 1 个分区 1 个备份。
-
3 个分区 1 个备份
创建一个主题为 one，创建3个分区，但仍为一个备份。 依次查看三台broker，可以看到每台 broker 中都有一个 one 主题的分区。
kakfos1 test-0
kakfos2 test-1
kakfos3 test-2 -
3 个分区 2 个备份
创建一个主题为 two，创建3个分区， 2个备份。 依次查看三台 broker，可以看到每台 broker 中都有2份 two 主题的分区。
kakfos1 test-0 test-2
kakfos2 test-1 test-0
kakfos3 test-2 test-1
4.2 查看分区与备份在 zk 中的信息
/brokers/ids：存放的是 kafka 集群中各个主机的 broker-id 列表。
get /brokers/ids/0 查看每个 id 的数据内容为当前主机的信息
/brokers/topics ：
/brokers/topics/city/partitions 中存放的是 city 主题下所包含的 partition。这里的 0、1、2，在/tmp/kafka-logs 目录中即为 city-0， city-1， city-2 。
get /brokers/topics/city/partitions/0/state 查看每个分区的信息。
4.3 segment
所谓的segment 其实就只在分区对应的文件夹下产生的文件。一个分区会被划分成大小相等的若干segment。这样一方面保证了分区的数据被划分到多个文件中保证不会产生体积过大的文件；另一方面可以基于这些segment文件进行历史数据的删除 提高效率。
一个segment又由 一个.log和一个.index文件组成，其中.log文件为数据文件用来存放数据分段数据 .index为索引文件保存对对应的.log文件的索引信息。这两个文件的命名规则为：partition全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值，数值大小为64位，20位数字字符长度，没有数字用0填充。
00000000000000000000.index 00000000000000000000.log
00000000000000170210.index 00000000000000170210.log
00000000000000239530.index 00000000000000239530.log
而在.index文件中 保存了对对应 .log文件的索引信息，通过查找.index文件可以获知每个存储在当前segment中的offset在.log文件中的开始位置。而每条日志有其固定格式，保存了包括offset编号、日志长度、key的长度等相关信息。通过这个固定格式中的数据可以确定出当前offset的结束位置 从而对数据进行读取。
真正开始读取指定分区中 某个offset对应的数据时 先根据offset和当前分区 的所有segment的名称做比较 确定出数据在哪个segment中，查找该segment的索引文件，确定当前offset在数据文件中的开始位置，从该位置开始读取数据文件，在根据数据格式判断结果，最终 获取到完整数据。
查看 segment
对于 segment 中的 log 文件，不能直接通过 cat 命令查找其内容，而是需要通过 kafka自带的一个工具查看。
bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
/tmp/kafka-logs/test-0/00000000000000000000.log --print-data-log
一个用户的一个主题会被提交到一个__consumer_offsets 分区中。 使用主题字符串的hash 值与 50 取模，结果即为分区索引 。
五、Kafka的可靠性
5.1 消费者消费数据时的可靠性保证
ISR， In-Sync Replicas，是指副本同步列表。
OSR， Outof-Sync Replicas，不在同步中的副本。
AR， Assigned ReplicasAR = ISR + ORS
在Kafka中维护了 一个AR列表(Assigned Replicas)，包括所有的分区的副本。
AR又分为ISR和OSR。只有ISR内的副本都同步了leader中的数据，该数据才能被提交，才能被消费者访问。OSR内的副本是否同步了leader的数据，不影响数据的提交，OSR内的follower 尽力的去同步leader，可能数据版本或落后。
最开始所有的副本都在ISR中，在kafka工作的过程中，如果某个副本同步速度慢于replica.lag.time.max.ms指定的阈值，则被踢出ISR 存入OSR；如果后续速度恢复可以回到ISR中。
如果都到OSR。看配置：严格模式，停止写服务。 非严格模式，选一个follower 当新leader(有可能丢数据)。
5.2 重复消费问题及解决方案
重复消费是同一个消息被同一个消费者/组多次消费。
最常见的重复消费有两种。
同一个 consumer 重复消费
offset
偏移量。 每条消息都有一个当前 Partition 下唯一的 64 字节的 offset，它是相对于当前分区第一条消息的偏移量。offset commit
Consumer 从 partition 中取出一批消息写入到 buffer 对其进行消费，在规定时间内消费完消息后，会自动将其消费消息的 offset 提交给 broker，以让 broker 记录下哪些消息是消费过的。当然，若在时限内没有消费完毕，其是不会提交 offset 的。提交的 offset 被写入到了一个特殊的主题__consumer_offsets 中。
kafka0.9 版本之前， offset 是由 zk 负责保存管理的，之后版本由 kafka broker 负责保存管理。
当 Consumer 由于消费能力较低而引发了消费超时时，则可能会形成重复消费。
在自动提交的情况下， 若消费者在超时时限内没有消费完毕读取来的这一批数据， 消费者会向 kafka 提交一个异常， 而不是 offset。 对于消费者来说， 由于其没有消费完毕， 所以其会再次向 kafka poll 消息消费。 而这次 poll 消息的 offset 还是原来的 offset， 所以这些消息就被重复消费了。
解决方案有两种：
- 延长自动提交的时限
- 自动提交改为手动提交
不同的 consumer 重复消费
当 Consumer 消费了消息但还未提交 offset 时宕机，则这些已被消费过的消息会被重复消费。
解决方案： 自动提交改为手动提交
5.3 HW和LEO
HW， HighWatermark，高水位，表示 Consumer 可以消费到的最高 partition 偏移量。 HW保证了 Kafka 集群中消息的一致性。确切地说，是在 broker 集群正常运转的状态下， 保证了partition 的 Follower 与 Leader 间数据的一致性。
LEO， Log End Offset，日志最新消息的偏移量。消息是被写入到 Kafka 的日志文件中的，这是当前最后一个写入的消息在 Partition 中的偏移量。
对于 leader 新写入的消息， consumer 是不能立刻消费的。 leader 会等待该消息被所有ISR 中的 partition follower 同步后才会更新 HW，此时消息才能被 consumer 消费。
AR ISR OSR LEO HW 这些信息都被保存在Zookeeper中。
5.4 截断机制
如果leader宕机 选出了新的leader 而新的leader并没有完全同步之前leader的所有数据，之后接受了后续新的数据，此时旧的leader恢复，则会发现新的leader中的数据和自己持有的数据不一致，此时旧的leader会将自己的数据截断到之前宕机之前的hw位置，之后同步新leader的数据。
如果ISR中的follower同步了leader中的部分数据，之后leader宕机，follower也宕机，此时选出新的leader可能同步了部分之前 leader的数据，之后接受新的数据，此时follower恢复过来，发现 自己持有的 数据和新 的leader的数据不一致，此时阶段数据到 之前的 hw位置，然后和新的leader同步数据。无法保证消息不丢失的。
老大死了。Follower抢做老大，截断机制，HW后的数据截断掉，再选老大。
旧老大活了，先截断，再与leader同步。
5.5 生产者发送消息的可靠性
生产者向leader发送数据时，可以选择需要的可靠性级别。
通过request.required.acks参数配置：
0 值
异步发送。生产者向 kafka 发送消息而不需要 kafka 反馈成功 ack。
影响：这种模式效率最高，可靠性最低。
可能在发送过程中，由于网络抖动等原因， 生产者发送的消息 kafka 并没有收到，丢失数据。
可能在leader宕机时丢失数据。
会存在消息丢失的情况。
1 值(默认值)
同步发送，默认值。生产者发送消息给 kafka， broker 的 partition leader 在收到消息后马上发送成功 ack（不需等待 ISRael 列表中的 Follower 同步完成的），生产者收到后知道消息发送成功，然后会再发送消息。如果一直未收到 kafka 的 ack，则生产者会认为消息发送失败，会重发消息。
问题 1： 该方式能否使生产者确认它发送的消息发送成功了？ 不能。
若生产者收到了 kafka 回复的 ack， 能否确认该消息已经成功被接收了？ 不能。答：若 partition leader 在接收到消息后， 其马上向 producer 回复 ack， 在 ISR 中的 follower还未同步该消息时， leader 挂了。 那么这条消息对于新的 Leader 来说， 根本就不存在， 即该消息丢失了。
问题 2： 该方法能否使生产者确认其发送的消息发送失败了？
答：能。 只要在超时时限内没有收到 ack， 那么该消息一定没有被 kafka 成功接收。
总结： 生产者接收到了 ack， 不能保证消息就一定被成功接收了， 但没有收到 ack， 一定是没有被成功接收的。 简单说就是“收到 Ack 不一定成功，没收到 ack 一定失败”。
-1 值（重复接收消息）
同步发送。 其值等同于 all。 生产者发送消息给 kafka， kafka 收到消息后，要等到 ISR 列表中的所有副本都同步消息完成后，才向生产者发送成功 ack。如果一直未收到 kafka 的 ack，则认为消息发送失败，会自动重发消息。
影响：
-
这种模式下可靠性很高，但是 当ISR列表中只剩下leader时，当leader宕机让然有可能丢数据。
此时可以配置min.insync.replicas指定要求观察ISR中至少要有指定数量的副本，默认该值为1，需要改为大于等于2的值。这样当生产者发送数据给leader但是发现ISR中只有leader自己时，会收到异常表明数据写入失败。
-
该方法可能会出现部分 follower 重复接收消息的情况（注意， 重复接收不同于重复消费）。
当 follower 还没有同步完成时， leader 挂了。 由于还没有同步完成， 所以 kafka 不会给生产者回复 ack， 所以生产者会重新发送消息给新的 leader。 若新的 Leader 曾同步过一部分原来的消息， 那么这些消息就是重复接收的消息。
5.6 Kafka可靠性的保证
- At most once: 消息可能会丢，但绝不会重复传输
- At least once：消息绝不会丢，但可能会重复传输
- Exactly once：每条消息肯定会被传输一次且仅传输一次
kafka最多保证At least once，可以保证不丢，但是可能会重复,为了解决重复需要引入唯一标识和去重机制，kafka提供了GUID实现了唯一标识，但是并没有提供自带的去重机制，需要开发人员基于业务规则自己去重。
做不到不多也不少，最多做到不丢但可能会多。
5.7 leader选举
当 leader 挂了后 broker controller 会从 ISR 中选一个 follower 成为新的 leader。 但，若 ISR
中的所有副本都挂了怎么办？ 可以通过 unclean.leader.election.enable 的取值来设置 Leader
选举的范围。
（1） false
默认值。 必须等待 ISR 列表中有副本活过来才进行新的选举。该策略可靠性有保证，但
可用性低。
（2） true
在 ISR 中没有副本的情况下可以选择任何一个没有宕机主机中该 topic 的 partition 副本
作为新的 leader，该策略可用性高，但可靠性没有保证。
六、kafka的优势
相比传统的消息系统，Kafka可以很好的保证有序性(1个分区)。
传统的队列在服务器上保存有序的消息，如果多个consumers同时从这个服务器消费消息，服务器就会以消息存储的顺序向consumer分发消息。虽然服务器按顺序发布消息，但是消息是被异步的分发到各consumer上，所以当消息到达时可能已经失去了原来的顺序，这意味着并发消费将导致顺序错乱。为了避免故障，这样的消息系统通常使用“专用consumer”的概念，其实就是只允许一个消费者消费消息，当然这就意味着失去了并发性。
在这方面Kafka做的更好，通过分区的概念，Kafka可以在多个consumer组并发的情况下提供较好的有序性和负载均衡。将每个分区分只分发给一个consumer组，这样一个分区就只被这个组的一个consumer消费，就可以顺序的消费这个分区的消息。因为有多个分区，依然可以在多个consumer组之间进行负载均衡。注意consumer组的数量不能多于分区的数量，也就是有多少分区就允许多少并发消费。
Kafka只能保证一个分区之内消息的有序性，在不同的分区之间是不可以的，这已经可以满足大部分应用的需求。如果需要topic中所有消息的有序性，那就只能让这个topic只有一个分区，当然也就只有一个consumer组消费它。
为什么大数据环境下的消息队列常选择kafka？
分布式存储数据，提供了更好的性能、可靠性、可扩展能力。
利用磁盘存储数据，且按照主题、分区来分布式存放数据，持久化存储，提供海量数据存储能力。
采用磁盘存储数据，连续进行读写保证性能，性能和磁盘的性能相关和数据量的大小无关。