一 Kafka 都有哪些特点？
- 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。
- 可扩展性：kafka集群支持热扩展
- 持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失
- 容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）
- 高并发：支持数千个客户端同时读写
二 Kafka的应用场景
- 日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、HBase、Solr等。
- 消息系统：解耦和生产者和消费者、缓存消息等。
- 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
- 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
- 流式处理：比如spark streaming和 Flink
三 Kafka 的设计架构
Kafka 架构分为以下几个部分
- Producer ：消息生产者，就是向 kafka broker 发消息的客户端。
- Consumer ：消息消费者，向 kafka broker 取消息的客户端。
- Topic ：可以理解为一个队列，一个 Topic 又分为一个或多个分区。
- Consumer Group：这是 kafka 用来实现一个 topic 消息的广播（发给所有的 consumer）和单播（发给任意一个 consumer）的手段。一个 topic 可以有多个 Consumer Group。
- Broker ：一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker 可以容纳多个 topic。
- Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker上，每个 partition 是一个有序的队列。partition 中的每条消息都会被分配一个有序的id（offset）。将消息发给 consumer，kafka 只保证按一个 partition 中的消息的顺序，不保证一个 topic 的整体（多个 partition 间）的顺序。
- Offset：kafka 的存储文件都是按照 offset.kafka 来命名，用 offset 做名字的好处是方便查找。例如你想找位于 2049 的位置，只要找到 2048.kafka 的文件即可。当然 the first offset 就是 00000000000.kafka。
- leader 是 partition的 leader ，leader 有 replication 数，Isr 机制，leader在broker 上，加入broker 挂掉，会在isr 里 找一个作为leader
ACK有三种
0
意味着producer不等待broker同步完成的确认，继续发送下一条(批)信息
提供了最低的延迟。但是最弱的持久性，当服务器发生故障时，就很可能发生数据丢失。例如leader已经死亡，producer不知情，还会继续发送消息broker接收不到数据就会数据丢失
1
意味着producer要等待leader成功收到数据并得到确认，才发送下一条message。此选项提供了较好的持久性较低的延迟性。
Partition的Leader死亡，follwer尚未复制，数据就会丢失
-1
意味着producer得到follwer确认，才发送下一条数据
每个partition不再只有一个，而是有一个leader(红色)和多个replica(蓝色)，生产者根据消息的topic和key值，确定了消息要发往哪个partition之后（假设是p1），会找到partition对应的leader(也就是broker2里的p1)，然后将消息发给leader，leader负责消息的写入，并与其余的replica进行同步。
ACK = 0 时 发送一次 不论leader是否接收
ACK = 1 时，等待leader接收成功即可
ACK = -1 时 ，需等待leader将消息同步给follower 简单来说就是，producer只有收到分区内所有副本的成功写入的通知才认为推送消息成功了。
综上所述，为了保证数据的可靠性，我们最少需要配置一下几个参数：
- producer 级别：acks=all（或者 request.required.acks=-1），同时发生模式为同步 producer.type=sync
- topic 级别：设置 replication.factor>=3，并且 min.insync.replicas>=2；
- broker 级别：关闭不完全的 Leader 选举，即 unclean.leader.election.enable=false；
四 Kafka 分区的目的？
分区对于 Kafka 集群的好处是：实现负载均衡。分区对于消费者来说，可以提高并发度，提高效率。
五 Kafka 是如何做到消息的有序性？
kafka 中的每个 partition 中的消息在写入时都是有序的，而且单独一个 partition 只能由一个消费者去消费，可以在里面保证消息的顺序性。但是分区之间的消息是不保证有序的。
六 Kafka 的再均衡
在Kafka中，当有新消费者加入或者订阅的topic数发生变化时，会触发Rebalance(再均衡：在同一个消费者组当中，分区的所有权从一个消费者转移到另外一个消费者)机制，Rebalance顾名思义就是重新均衡消费者消费。Rebalance的过程如下：
第一步：所有成员都向coordinator发送请求，请求入组。一旦所有成员都发送了请求，coordinator会从中选择一个consumer担任leader的角色，并把组成员信息以及订阅信息发给leader。
第二步：leader开始分配消费方案，指明具体哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案发给coordinator。coordinator接收到分配方案之后会把方案发给各个consumer，这样组内的所有成员就都知道自己应该消费哪些分区了。
所以对于Rebalance来说，Coordinator起着至关重要的作用
七 Kafka 在什么情况下会出现消息丢失
- 数据的一致性和可靠性
- 可靠性 是 通过分区 和 副本 ISR ，和ack机制 来保证数据的可靠性
- 一致性 是 High Water Mark 取决于 ISR 列表里面偏移量最小的分区，对应于上图的副本2，这个很类似于木桶原理。
消费者 只能看到 Message2 这条消息，保证了数据的一致性。
Kafka 复制机制
Kafka 主题中的每个分区都有一个预写日志（write-ahead log），我们写入 Kafka 的消息就存储在这里面。这里面的每条消息都有一个唯一的偏移量，用于标识它在当前分区日志中的位置。如下图所示：
Kafka 中的每个主题分区都被复制了 n 次，其中的 n 是主题的复制因子（replication factor）。这允许 Kafka 在集群服务器发生故障时自动切换到这些副本，以便在出现故障时消息仍然可用。Kafka 的复制是以分区为粒度的，分区的预写日志被复制到 n 个服务器。 在 n 个副本中，一个副本作为 leader，其他副本成为 followers。顾名思义，producer 只能往 leader 分区上写数据（读也只能从 leader 分区上进行），followers 只按顺序从 leader 上复制日志。
日志复制算法（log replication algorithm）必须提供的基本保证是，如果它告诉客户端消息已被提交，而当前 leader 出现故障，新选出的 leader 也必须具有该消息。在出现故障时，Kafka 会从挂掉 leader 的 ISR 里面选择一个 follower 作为这个分区新的 leader ；换句话说，是因为这个 follower 是跟上 leader 写进度的。
每个分区的 leader 会维护一个 in-sync replica（同步副本列表，又称 ISR）。当 producer 往 broker 发送消息，消息先写入到对应 leader 分区上，然后复制到这个分区的所有副本中。只有将消息成功复制到所有同步副本（ISR）后，这条消息才算被提交。由于消息复制延迟受到最慢同步副本的限制，因此快速检测慢副本并将其从 ISR 中删除非常重要。 Kafka 复制协议的细节会有些细微差别。
副本在什么情况下才算跟上 leader
一个副本如果它没有跟上 leader 的日志进度，那么它可能会被标记为不同步的副本。我通过一个例子来解释跟上（caught up）的含义。假设我们有一个名为 foo 的主题，并且只有一个分区，同时复制因子为 3。假设此分区的副本分别在 brokers 1，2和3上，并且我们已经在主题 foo 上提交了3条消息。brokers 1上的副本是 leader，副本2和3是 followers，所有副本都是 ISR 的一部分。假设 replica.lag.max.messages 设置为4，这意味着只要 follower 落后 leader 的消息不超过3条，它就不会从 ISR 中删除。我们把 replica.lag.time.max.ms 设置为500毫秒，这意味着只要 follower 每隔500毫秒或更早地向 leader 发送一个 fetch 请求，它们就不会被标记为死亡并且不会从 ISR 中删除。
现在假设 producer 往 leader 上发送下一条消息，与此同时，broker 3 上发生了 GC 停顿，现在每个 broker 上的分区情况如下所示：
由于 broker 3 在 ISR中，因此在将 broker 3从 ISR 中移除或 broker 3 上的分区跟上 leader 的日志结束偏移之前，最新消息都是不认为被提交的。注意，由于 border 3 落后 leader 的消息比 replica.lag.max.messages = 4 要小，因此不符合从 ISR 中删除的条件。这意味着 broker 3 上的分区需要从 leader 上同步 offset 为 3 的消息，如果它做到了，那么这个副本就是跟上 leader 的。假设 broker 3 在 100ms 内 GC 完成了，并且跟上了 leader 的日志结束偏移，那么最新的情况如下图：
什么情况下会导致一个副本与 leader 失去同步
一个副本与 leader 失去同步的原因有很多，主要包括：
- 慢副本（Slow replica）：follower replica 在一段时间内一直无法赶上 leader 的写进度。造成这种情况的最常见原因之一是 follower replica 上的 I/O瓶颈，导致它持久化日志的时间比它从 leader 消费消息的时间要长；
- 卡住副本（Stuck replica）：follower replica 在很长一段时间内停止从 leader 获取消息。这可能是以为 GC 停顿，或者副本出现故障；
- 刚启动副本（Bootstrapping replica）：当用户给某个主题增加副本因子时，新的 follower replicas 是不同步的，直到它跟上 leader 的日志。
当副本落后于 leader 分区时，这个副本被认为是不同步或滞后的。在 Kafka 0.8.2 中，副本的滞后于 leader 是根据 replica.lag.max.messages 或 replica.lag.time.max.ms 来衡量的； 前者用于检测慢副本（Slow replica），而后者用于检测卡住副本（Stuck replica）。
如何确认某个副本处于滞后状态
通过 replica.lag.time.max.ms 来检测卡住副本（Stuck replica）在所有情况下都能很好地工作。它跟踪 follower 副本没有向 leader 发送获取请求的时间，通过这个可以推断 follower 是否正常。另一方面，使用消息数量检测不同步慢副本（Slow replica）的模型只有在为单个主题或具有同类流量模式的多个主题设置这些参数时才能很好地工作，但我们发现它不能扩展到生产集群中所有主题。
在我之前的示例的基础上，如果主题 foo 以 2 msg/sec 的速率写入数据，其中 leader 收到的单个批次通常永远不会超过3条消息，那么我们知道这个主题的 replica.lag.max.messages 参数可以设置为 4。为什么？ 因为我们以最大速度往 leader 写数据并且在 follower 副本复制这些消息之前，follower 的日志落后于 leader 不超过3条消息。同时，如果主题 foo 的 follower 副本始终落后于 leader 超过3条消息，则我们希望 leader 删除慢速 follower 副本以防止消息写入延迟增加。
这本质上是 replica.lag.max.messages 的目标 - 能够检测始终与 leader 不同步的副本。假设现在这个主题的流量由于峰值而增加，生产者最终往 foo 发送了一批包含4条消息，等于 replica.lag.max.messages = 4 的配置值。此时，两个 follower 副本将被视为与 leader 不同步，并被移除 ISR。
但是，由于两个 follower 副本都处于活动状态，因此它们将在下一个 fetch 请求中赶上 leader 的日志结束偏移量并被添加回 ISR。如果生产者继续向 leader 发送大量的消息，则将重复上述相同的过程。这证明了 follower 副本进出 ISR 时触发不必要的错误警报的情况。
replica.lag.max.messages 参数的核心问题是，用户必须猜测如何配置这个值，因为我们不知道 Kafka 的传入流量到底会到多少，特别是在网络峰值的情况下。
一个参数搞定一切
我们意识到，检测卡住或慢速副本真正重要的事情，是副本与 leader 不同步的时间。我们删除了通过猜测来设置的 replica.lag.max.messages 参数。现在，我们只需要在服务器上配置 replica.lag.time.max.ms 参数即可；这个参数的含义为副本与 leader 不同步的时间。
- 检测卡住副本（Stuck replica）的方式与以前相同 - 如果副本未能在
replica.lag.time.max.ms时间内发送 fetch 请求，则会将其视为已死的副本并从 ISR 中删除； - 检测慢副本的机制已经改变 - 如果副本落后于 leader 的时间超过
replica.lag.time.max.ms，则认为它太慢并且从 ISR 中删除。
因此，即使在峰值流量下，生产者往 leader 发送大量的消息，除非副本始终和 leader 保持 replica.lag.time.max.ms 时间的落后，否则它不会随机进出 ISR。
常见问题（ISR）
Leader丢失导致的Consumer挂起故障解决
当时的现象是Leader的值均为-1，Isr的值也均为-1
至此我们猜测是因为KakfaController丢失导致的partition leader为-1，进而导致的Consumer端无法正常消费。
我们可以发现Replica的状态机管理是在KafkaController中完成的，也就是说Controller丢失的情况下，也就失去了与Zookeeper交互的能力。默认情况下Leader必须从ISR列表中选择，我们发现列表为空（经过排查发现是Kafka的bug，在Controller和Zookeeper通信过程中出现问题时，可能导致leader丢失而无法通信的情况，这个可能性是很大的，因为zookeeper在高并发环境是容易超时，这就是为什么在kafka 0.8.2.1之后更建议我们使用kafka topic的方式存储offset而不是存储在zookeeper中。
kafka broker Leader -1引起消费异常
Kafka Broker Leader 为-1，表示有partition在选举Leader的时候失败了，因此引起了消费该Topic的实时任务都出现了异常，经过排除发现挂掉的cdh-003机器正好是broker id为257。(但为何192没有被选举为leader呢？？)
192没选举为leader的原因是选择了unclean.leader.election.enable=false，即非ISR不能参与leader选举
kafka producer生产数据server端堆外内存频繁OOM
先调发送缓存大小。
max.block.ms之后，抛出异常。此设置应该大致的对应生产者将要使用的总内存，但不是硬约束，因为生产者所使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启动压缩），以及用于保持发送中的请求。
比如 Topic: __consumer_offsets Partition: 8 Leader: 2 Replicas: 2,4,1,3 Isr: 2
isr 只有一个 2, 一旦 2 节点掉了,是不是部分数据就没办法访问
root@zx001 kafka]# su kafka
[kafka@zx001 kafka]$ bin/kafka-topics.sh --zookeeper zx008:2181 --topic __consumer_offsets --describe
Topic:__consumer_offsets PartitionCount:50 ReplicationFactor:4 Configs:segment.bytes=104857600,cleanup.policy=compact,compression.type=uncompressed
Topic: __consumer_offsets Partition: 0 Leader: 2 Replicas: 2,1,3,4 Isr: 2
Topic: __consumer_offsets Partition: 1 Leader: 3 Replicas: 3,2,4,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 2 Leader: 4 Replicas: 4,3,1,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 3 Leader: 1 Replicas: 1,4,2,3 Isr: 1,3,4,2
Topic: __consumer_offsets Partition: 4 Leader: 2 Replicas: 2,3,4,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 5 Leader: 3 Replicas: 3,4,1,2 Isr: 1,3,4,2
Topic: __consumer_offsets Partition: 6 Leader: 4 Replicas: 4,1,2,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 7 Leader: 1 Replicas: 1,2,3,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 8 Leader: 2 Replicas: 2,4,1,3 Isr: 2
Topic: __consumer_offsets Partition: 9 Leader: 3 Replicas: 3,1,2,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 10 Leader: 4 Replicas: 4,2,3,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 11 Leader: 1 Replicas: 1,3,4,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 12 Leader: 2 Replicas: 2,1,3,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 13 Leader: 3 Replicas: 3,2,4,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 14 Leader: 4 Replicas: 4,3,1,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 15 Leader: 1 Replicas: 1,4,2,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 16 Leader: 2 Replicas: 2,3,4,1 Isr: 1,3,4,2
Topic: __consumer_offsets Partition: 17 Leader: 3 Replicas: 3,4,1,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 18 Leader: 4 Replicas: 4,1,2,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 19 Leader: 1 Replicas: 1,2,3,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 20 Leader: 2 Replicas: 2,4,1,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 21 Leader: 3 Replicas: 3,1,2,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 22 Leader: 4 Replicas: 4,2,3,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 23 Leader: 1 Replicas: 1,3,4,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 24 Leader: 2 Replicas: 2,1,3,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 25 Leader: 3 Replicas: 3,2,4,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 26 Leader: 4 Replicas: 4,3,1,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 27 Leader: 1 Replicas: 1,4,2,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 28 Leader: 2 Replicas: 2,3,4,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 29 Leader: 3 Replicas: 3,4,1,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 30 Leader: 4 Replicas: 4,1,2,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 31 Leader: 1 Replicas: 1,2,3,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 32 Leader: 2 Replicas: 2,4,1,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 33 Leader: 3 Replicas: 3,1,2,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 34 Leader: 4 Replicas: 4,2,3,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 35 Leader: 1 Replicas: 1,3,4,2 Isr: 1,3,4,2
Topic: __consumer_offsets Partition: 36 Leader: 2 Replicas: 2,1,3,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 37 Leader: 3 Replicas: 3,2,4,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 38 Leader: 4 Replicas: 4,3,1,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 39 Leader: 1 Replicas: 1,4,2,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 40 Leader: 2 Replicas: 2,3,4,1 Isr: 2
Topic: __consumer_offsets Partition: 41 Leader: 3 Replicas: 3,4,1,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 42 Leader: 4 Replicas: 4,1,2,3 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 43 Leader: 1 Replicas: 1,2,3,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 44 Leader: 2 Replicas: 2,4,1,3 Isr: 2
Topic: __consumer_offsets Partition: 45 Leader: 3 Replicas: 3,1,2,4 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 46 Leader: 4 Replicas: 4,2,3,1 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 47 Leader: 1 Replicas: 1,3,4,2 Isr: 1,4,3,2
Topic: __consumer_offsets Partition: 48 Leader: 2 Replicas: 2,1,3,4 Isr: 2
Topic: __consumer_offsets Partition: 49 Leader: 3 Replicas: 3,2,4,1 Isr: 1,4,3,2
[kafka@zx001 kafka]$
个人的猜测原因是，最后leader 和 isr 都只剩一个了，且leader和isr的broker id是一样，但是后续leader 也出问题了，所以显示的isr有broker，其实是最后控制partition分区的leader的broker id，而且最后因为一些原因也挂掉了，ISR也停留在了最后一个leader id上。
如何确认不重复消费和丢数据的话，生产者用同步消息确认，消费者用手动提交offset
消费者无法消费数据
今天特意研究了一下这个问题，专门来登录回答一下楼主的疑问：
其实消费者无法消费有如下两种情况：
1）就是consumer_offsets副本数为1， 当你所使用的消费者分组所对应的consumer_offsets分区所在的broker挂掉了的时候，消费者就无法读取offset和提交offset了，这就是楼主所分析的这种现象，通过根据集群规模修改offsets.topic.replication.factor大于1即可解决。
2）就是楼主的“未解之谜”， 当你在未创建topic的情况下直接使用kafka-console-producer.sh创建生产者的时候，所创建的topic的默认分区数为1，副本数为1， 也就是说此时此topic的所有分区和副本都只存在于一个kafka broker上（一般都是第一个启动的broker，具体源码还没来得及看），此时，无论你所用的消费者所属的consumer_offsets分区再哪个broker上，只要broker1挂了，生产者和消费者都不可用，因为topic所在分区挂了，生产者肯定无法写数据了，那么消费者需要从topic分区fetch数据，肯定 也不行了。
如果，topic分区副本和消费者组所属的consumer-offsets不在一个broker上的话，此时，如果topic分区所在的broker挂了，那么生产者无法使用，如果消费者组所属的__consumer-offsets分区所在的broker挂了，那么消费者不可用。
经过我的反复测试，在三节点的集群中，只要把所使用的topic副本数设置为大于1即可，offsets.topic.replication.factor大于1即可解决这种问题！
一个消费者组消费消息的offset只会存在__consumer_offsets 里的一个分区上面，所以如果副本只有一份，那就只会在特定的broker上
最近碰到一个问题，多个业务往向一个Kafka topic发送消息，有些业务的消费量很大，有些业务的消息量很小。因Kafka尚未较好地支持按优先级来消费消息，导致某些业务的消息消费延时的问题。一种简单的解决方案是再增加几个Topic，面对一些系统遗留问题，增加Topic带来的是生产者和消费者处理逻辑复杂性。一种方法是使用Kafka Standalone consumer，先使用consumer.partitionFor("TOPIC_NAME")获取topic下的所有分区信息，再使用consumer.assign(partitions)显示地为consumer指定消费分区。另一种方法是基于consumer group 自定义Kafka consumer的分区分配策略，那这时候就得对Kafka目前已有的分区分配策略有所了解，并且明白什么时候、什么场景下触发rebalance？
Kafka consumer要消费消息，哪些的分区的消息交给哪个consumer消费呢？这是consumer的分区分配策略，默认有三个：range、round-robin、sticky。说到round-robin这个算法，真是无处不在，它经常用在一些需要负载均衡的场景。比如Elasticsearch client向ES Server发送搜索请求时，因为默认情况下每台ES节点都可做为coordinator节点接收用户的查询请求，而在coordinator节点上需要汇总所有分片的查询结果，这需要消耗大量的内存和CPU，因此ES Client 也是基于round-robin算法选择将查询请求发送到哪个ES节点上。如果你仔细留意，会发现在Redis里面也会有这个算法的身影。再比如说：Redis LRU Cache中关于Key的access pattern，一般服从幂指分布(power-law distribution)：具有某一特征的一小部分的Key访问频率远远大于其他的Key，正如这种访问特性，LRU能达到很好的缓存效果。另外，Redis sorted set类型是基于skiplist实现，新的skipNode节点最大层数设置为多少合适呢？这也是个power-law distribution问题，其源码注释中：
Returns a random level for the new skiplist node we are going to create. The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL (both inclusive), with a powerlaw-alike distribution where higher levels are less likely to be returned.
其实，我想表达的是有些思想或者说是解决方案，它是通用的，应用于各个不同的存储系统中，将它们对比起来看，能更好地理解系统背后的原理。
最近每次想写一些笔记时，脑海里总是出现一些其他各种各样的想法。这次本来主要是想写kafka 中这两个配置参数：session.timeout.ms 和 heartbeat.interval.ms的区别的，结果就先扯了一通数据存储相关的东西。
下面继续：
因为一个topic往往有多个分区，而我们又会在一个consumer group里面创建多个消费者消费这个topic，因此：就有了一个问题：哪些的分区的消息交给哪个consumer消费呢？这里涉及到三个概念：consumer group，consumer group里面的consumer，以及每个consumer group有一个 group coordinator。conusmer分区分配是通过组管理协议来实施的：具体如下：
consumer group里面的各个consumer都向 group coordinator发送JoinGroup请求，这样group coordinator就有了所有consumer的成员信息，于是它从中选出一个consumer作为Leader consumer，并告诉Leader consumer说：你拿着这些成员信息和我给你的topic分区信息去安排一下哪些consumer负责消费哪些分区吧
接下来，Leader consumer就根据我们配置的分配策略(由参数partition.assignment.strategy指定)为各个consumer计算好了各自待消费的分区。于是，各个consumer向 group coordinator 发送SyncGroup请求，但只有Leader consumer的请求中有分区分配策略，group coordinator 收到leader consumer的分区分配方案后，把该方案下发给各个consumer。画个图，就是下面这样的：
而在正常情况下 ，当有consumer进出consumer group时就会触发rebalance，所谓rebalance就是重新制订一个分区分配方案。而制订好了分区分配方案，就得及时告知各个consumer，这就与 heartbeat.interval.ms参数有关了。具体说来就是：每个consumer 都会根据 heartbeat.interval.ms 参数指定的时间周期性地向group coordinator发送 hearbeat，group coordinator会给各个consumer响应，若发生了 rebalance，各个consumer收到的响应中会包含 REBALANCE_IN_PROGRESS 标识，这样各个consumer就知道已经发生了rebalance，同时 group coordinator也知道了各个consumer的存活情况。
那为什么要把 heartbeat.interval.ms 与 session.timeout.ms 进行对比呢？session.timeout.ms是指：group coordinator检测consumer发生崩溃所需的时间。一个consumer group里面的某个consumer挂掉了，最长需要 session.timeout.ms 秒检测出来。举个示例session.timeout.ms=10，heartbeat.interval.ms=3
session.timeout.ms是个"逻辑"指标，它指定了一个阈值---10秒，在这个阈值内如果coordinator未收到consumer的任何消息，那coordinator就认为consumer挂了。而heartbeat.interval.ms是个"物理"指标，它告诉consumer要每3秒给coordinator发一个心跳包，heartbeat.interval.ms越小，发的心跳包越多，它是会影响发TCP包的数量的，产生了实际的影响，这也是我为什么将之称为"物理"指标的原因。
如果group coordinator在一个heartbeat.interval.ms周期内未收到consumer的心跳，就把该consumer移出group，这有点说不过去。就好像consumer犯了一个小错，就一棍子把它打死了。事实上，有可能网络延时，有可能consumer出现了一次长时间GC，影响了心跳包的到达，说不定下一个heartbeat就正常了。
而heartbeat.interval.ms肯定是要小于session.timeout.ms的，如果consumer group发生了rebalance，通过心跳包里面的REBALANCE_IN_PROGRESS，consumer就能及时知道发生了rebalance，从而更新consumer可消费的分区。而如果超过了session.timeout.ms，group coordinator都认为consumer挂了，那也当然不用把 rebalance信息告诉该consumer了。
在kafka0.10.1之后的版本中，将session.timeout.ms 和 max.poll.interval.ms 解耦了。也就是说：new KafkaConsumer对象后，在while true循环中执行consumer.poll拉取消息这个过程中，其实背后是有2个线程的，即一个kafka consumer实例包含2个线程：一个是heartbeat 线程，另一个是processing线程，processing线程可理解为调用consumer.poll方法执行消息处理逻辑的线程，而heartbeat线程是一个后台线程，对程序员是"隐藏不见"的。如果消息处理逻辑很复杂，比如说需要处理5min，那么 max.poll.interval.ms可设置成比5min大一点的值。而heartbeat 线程则和上面提到的参数 heartbeat.interval.ms有关，heartbeat线程 每隔heartbeat.interval.ms向coordinator发送一个心跳包，证明自己还活着。只要 heartbeat线程 在 session.timeout.ms 时间内 向 coordinator发送过心跳包，那么 group coordinator就认为当前的kafka consumer是活着的。
在kafka0.10.1之前，发送心跳包和消息处理逻辑这2个过程是耦合在一起的，试想：如果一条消息处理时长要5min，而session.timeout.ms=3000ms，那么等 kafka consumer处理完消息，group coordinator早就将consumer 移出group了，因为只有一个线程，在消息处理过程中就无法向group coordinator发送心跳包，超过3000ms未发送心跳包，group coordinator就将该consumer移出group了。而将二者分开，一个processing线程负责执行消息处理逻辑，一个heartbeat线程负责发送心跳包，那么：就算一条消息需要处理5min，只要底heartbeat线程在session.timeout.ms向group coordinator发送了心跳包，那consumer可以继续处理消息，而不用担心被移出group了。另一个好处是：如果consumer出了问题，那么在 session.timeout.ms内就能检测出来，而不用等到 max.poll.interval.ms 时长后才能检测出来。
一次kafka consumer 不断地 rebalance 分析
明白了session.timeout.ms 和 max.poll.interval.ms 和 heartbeat.interval.ms三个参数的意义后，现在来实际分析一下项目中经常碰到的 consumer rebalance 错误。
一般我们是在一个线程（用户线程）里面执行kafka consumer 的while true循环逻辑的，其实这里有2个线程：一个是用户线程，另一个是心跳线程。心跳线程，我想就是根据heartbeat.interval.ms参数配置的值周期性向coordinator发送心跳包以证明consumer还活着。
如果消息处理逻辑过重，也即用户线程需要执行很长的时间处理消息，然后再提交offset。咋一看，有一个后台心跳线程在不断地发送心跳啊，那为什么group coordinator怎么还老是将consumer移出group，然后导致不断地rebalance呢？
我想，问题应该是 max.poll.interval.ms这个参数引起的吧，因为在ERROR日志中，老是提示：消息处理逻辑花了太长的时间，要么减少max.poll.records值，要么增大session.timeout.ms的值。尽管有后台heartbeat 线程，但是如果consumer的消息处理逻辑时长超过了max.poll.interval.ms ，那么此consumer提交offset就会失败：
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.
此外，在用户线程中，一般会做一些失败的重试处理。比如通过线程池的 ThreadPoolExecutor#afterExecute()方法捕获到异常，再次提交Runnable任务重新订阅kafka topic。本来消费处理需要很长的时间，如果某个consumer处理超时：消息处理逻辑的时长大于max.poll.interval.ms (或者消息处理过程中发生了异常)，被coordinator移出了consumer组，这时由于失败的重试处理，自动从线程池中拿出一个新线程作为消费者去订阅topic，那么意味着有新消费者加入group，就会引发 rebalance，而可悲的是：新的消费者还是来不及处理完所有消息，又被移出group。如此循环，就发生了不停地 rebalance 的现象。
kafka的rebalance机制
在Kafka中，当有新消费者加入或者订阅的Topic数发生变化时，会触发Rebalance(再均衡：在同一个消费者组当中，分区的所有权从一个消费者转移到另外一个消费者)机制，Rebalance顾名思义就是重新均衡消费者消费。Rebalance的过程如下：
第一步：所有消费成员都向Coordinator发送请求，请求入Consumer Group。一旦所有成员都发送了请求，Coordinator会从中选择一个Consumer担任Leader的角色，并把组成员信息以及订阅信息发给Leader。
第二步：Leader开始分配消费方案，指明具体哪个Consumer负责消费哪些Topic的哪些Partition。一旦完成分配，leader会将这个方案发给Coordinator。Coordinator接收到分配方案之后会把方案发给各个Consumer，这样组内的所有成员就都知道自己应该消费哪些分区了。
所以对于Rebalance来说，Coordinator起着至关重要的作用
二、rebalance可能发生的时机
1、分区个数的增加
2、对Topic的订阅发生变化
3、消费组成员的加入或离开（这个是我们最常遇到）
三、rebalance的影响
Rebalance对我们数据的影响主要有以下几点：
1、可能重复消费: Consumer被踢出消费组，可能还没有提交offset，Rebalance时会Partition重新分配其它Consumer,会造成重复消费，虽有幂等操作但耗费消费资源，亦增加集群压力
2、集群不稳定：Rebalance扩散到整个ConsumerGroup的所有消费者，因为一个消费者的退出，导致整个Group进行了Rebalance，并在一个比较慢的时间内达到稳定状态，影响面较大
3、影响消费速度：频繁的Rebalance反而降低了消息的消费速度，大部分时间都在重复消费和Rebalance
四、避免rebalance措施
1、业务需要不可避免
（1）针对分区个数的增加， 一般不会常有，是需要增加的时候都是业务及数据需求，不可避免
（2）对Topic的订阅增加或取消亦不可避免
2、合理设置消费者参数
下边是我们遇到的，要格外关注及重视
（1）未能及时发送心跳而Rebalance
session.timeout.ms 一次session的连接超时时间
heartbeat.interval.ms 心跳时间，一般为超时时间的1/3，Consumer在被判定为死亡之前，能够发送至少 3 轮的心跳请求
（2）Consumer消费超时而Rebalance
max.poll.interval.ms 每隔多长时间去拉取消息。合理设置预期值，尽量但间隔时间消费者处理完业务逻辑，否则就会被coordinator判定为死亡，踢出Consumer Group，进行Rebalance
max.poll.records 一次从拉取出来的数据条数。根据消费业务处理耗费时长合理设置，如果每次max.poll.interval.ms 设置的时间较短，可以max.poll.records设置小点儿，少拉取些，这样不会超时。
总之，尽可能在max.poll.interval.ms时间间隔内处理完max.poll.records条消息，让Coordinator认为消费Consumer还活着
1 flume的source是kafka，sink是hdfs，怎样判断flume是否堆积，或者是说怎么样保证落地的速度和消费的速度是平衡的
2 怎么判断flume的agent程序是否挂掉
3 挂掉时tmp文件爱呢怎么处理（hdfs上的tmp文件）
4 我遇到一个问题，当agent是6个时，一小时约生成26.5G文件，当有3个agent时，一小时16.9G，按我的理解，1个agent相当于一个consumer，这6个agent是同一个consumer group，那么3个和6个agent1小时都应该生成那么多的文件，但实际不是，怎么回事？
先说第4个问题，其实同事理解的不错，kafka的consumer必须归于一个group，以group为单位去消费topic上的数据，group内的consumer不论多或少，最终输出的数据量按理应该是一样多的。至于为啥当consumer多时消费量多，consumer少时消费少，那就说明了，每个consumer都到了它所能处理的上限了，而topic中整体的数据量太大。我的建议是增大consumer的数量，果然当增大consumer的数量后，消费的总量也上去了，说明consumer的数量还有提高的空间，当然，不论消费的多少，数据是不会丢的
再来说第一个问题，因为数据量很大了，conusmer的处理能力达到上限，然后flume作为kafka 的consumer，出现了消息的积压（这哥们在flume中加了个interceptor），按理说，我flume消费多少，我就从kafka中pull多少，根据我的消费能力我来消费数据，如果真是这样就不会有堆积了，kafkasource 默认batchsize 是1000条数据，而且auto.commit.enable 可能设置为true，所以，在flume往channel写的中间，一个interceptor处理能力上不去的话，kafkasource不停的pull数据过来，就导致了数据的来不及处理，有积压的出现。我的理解kafkasource 和 interceptor 之间或许没有建立一种ack机制，kafkasource 根据自己的batchsize和commit设置去不断的拉取数据，在interceptor处出现了瓶颈，如果source和interceptor能统一起来，就解决问题了。或者治标的办法是，降低batchsize 的大小，设置为500，auto.commit.enable = false 。在没改设置之前，出现了kafkasource的内存溢出，报的是由于GC导致的OOM，我分析了他自定义的interceptor代码，发现，它在event processor处理方法内，new 对象过多，对代码进行了优化后，问题暂时解决。治本的方法就是在可以在batch event processor中加上offset的处理，例如在init方法中初始化一个zookeeper的客户端，在batch event processor中对offset更新到zookeeper中，或者是更新到kafka的一个topic中。kafkasource根据offset来pull数据（或者是ack机制，当处理完这一批数据后，ack后，kafkasource再去拉取下一批数据，当数据没有处理成功的话，超时的话，也直接拉取下一批数据，有数据丢失的情况）
第二个问题，如何判断agent挂掉，可以在interceptor中做文章，例如在init方法中初始化一个zookeeper客户端，然后在zk上注册一个watch，当agent挂掉后，interceptor也就挂掉了，zookeeper就能监测到。
第三个问题，tmp文件是因为当sink往hdfs写的过程中，正常写入的文件被命名为tmp文件，当agent当掉后，tmp不会自动改名字，但是tmp中也会有部分数据存在，所以可以写个程序将文件名中的tmp去掉。如果agent重启后，可以续传，就保证数据连接上了。
前面说过了，同事加了一个interceptor后，会导致flume的性能下降。
同时有报了如下的错误
org.apache.flume.source.kafka.KafkaSource.doProcess(KafkaSource.java:314)] KafkaSource EXCEPTION, {}
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:552)
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:493)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:665)
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:644)
我的理解是interceptor，当时处理时间过长，consumer 等待下一次pull的时间间隔过长，导致发送hearbeat时间间隔过长，coordinator认为comsumer死了，就发生了rebalance。
增大参数 hearbeat.interval.ms
缩小参数 max.partition.fetch.bytes 让每次pull的数据少点
新增多个 Flume 实例后，Kafka 数据重复消费问题处理
我们使用 Flume 将数据从 Kafka 加载到 Hive 中。
（自己）假如启动2个agent消费，会触发 reblance，假如有10个分区，flume 会自动建立10个consumer，1个groupid，如果在启动一个，触发relance，2个flume agent 平分 kafka的partition ，也就是一个5个。reblance 一般会发生在消费组 这端，生产者 主要是isr问题。
由于启动一个 Flume 实例时，数据加载的速度只能达到 10MB/秒 (每条Kafka记录100B)。于是我们计划启动多个 Flume 实例 (指定同一个消费者组名称)。
我们知道 Kafka 数据消费是以 Partition 为单位的，即一个 Partition 只能被一个 Flume 实例消费。当启动第二个 Flume 实例时，Kafka 会把已经分配给第一个 Flume 实例的 Partition 回收(revoke)后，然后重新分配(完整过程在这里 https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Client-side+Assignment+Proposal )。
另一方面，Flume 为了满足事务语义，需要等每条 Kafka Record 真正放到 Channel 后，才能向 Kafka 确认 (调用 consumer.commitSync)。 在 Kafka 回收，分配过程 (rebalance) 中，已经接收到 Flume 端但还未被 Flume 放到 Channel 的 Records (自然也就没有向 Kafka 确认) 就被Kafka 认为未被消费；特别是属于将来被分配给其他 Flume 实例的 Partition 上的 Records，就会再次被新的 Flume 实例消费，造成重复加载。
Kafka 提供 ConsumerRebalanceListener 接口允许消费者侦听这个回收(onPartitionsRevoked)、分配(onPartitionsAssigned)过程。Flume 中 Kafka Source 在侦测到rebalance时的代码如下:
// this flag is set to true in a callback when some partitions are revoked.
// If there are any records we commit them.
if (rebalanceFlag.get()) {
rebalanceFlag.set(false);
break;
}
这个措施很简单，遇到这种情况立即跳出去处理(放到 Channel 并向 kafka 确认)。从代码 (https://github.com/apache/flume/blob/flume-1.8/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java) 上下文可以看到这里有几个缺陷:
1. 由于把 event 放到 channel 是需要时间的 (执行 interceptor 和 channel selector)，不一定能够在 partition 分配给别的 Flume 实例之前完成并提交 Kafka (调研 consumer.commitSync)
2. 那些已经读取到 records (本地) 中，但是没有放到 eventList 中的记录没有处理
所以我们修改了 Flume 源码，对这两个缺陷进行补救:
// If there are any records we commit them.
if (rebalanceFlag.get()) {
rebalanceFlag.set(false);
// invalidate remaining records
it = null;
// and drop processed events
eventList.clear();
tpAndOffsetMetadata.clear();
// and seek to committed offsets
for (Map.Entry<String, List<PartitionInfo>> es : consumer.listTopics().entrySet()) {
for (PartitionInfo pi : es.getValue()) {
TopicPartition tp = new TopicPartition(pi.topic(), pi.partition());
try {
OffsetAndMetadata oam = consumer.committed(tp);
if (oam != null) {
consumer.seek(tp, oam.offset());
}
} catch (Exception e) {
// log.warn("ignore seeking exception, {}", e);
}
}
}
log.info("read-ahead records have been dropped.");
break;
}
详细请参考 https://github.com/hejiang2000/flume/blob/hejiang-kafka-source/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java 。
经测试，以上处理完全修正 Kafka Source 数据重复消费问题。
————————————————
版权声明：本文为CSDN博主「GBASE-何江」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/hejiangtju/article/details/80364388
理解 ：意思是忽略已经加入到eventList里面的消息，重新设置seek，然后重新读取一遍
阿里云的文档
使用消息队列Kafka版时消费客户端频繁出现Rebalance
操作系统优化
大部分Linux发布版本默认的内核参数配置能让大部分应用工作的相当好。但对于实际的Kafka broker场景来说，做稍些改变会提升broker性能。主要涉及的配置：虚拟内存、网络和磁盘挂载（用来存储log segment），一般在 /etc/sysctl.conf (CentOS系统)。
Virtual Memory
一般来说，Linux的虚拟内存会根据系统负载自动调整。内存页（page）swap到磁盘会显著的影响Kafka的性能，并且Kafka重度使用page cache，如果VM系统swap到磁盘，那说明没有足够的内存来分配page cache。
避免swap的一种方式是设置swap空间为0。但是，swap会在系统崩溃时提供安全机制，或者会在out of memory的情况下阻止操作系统 kill 掉进程。由于这个原因，推荐 vm.swappiness 参数设置为一个非常低的值：1 。这个参数表示 VM系统中的多少百分比用来作为swap空间。
另外一种方式是通过内核调节“脏页”（注：“脏页”会被刷到磁盘上）。Kafka依赖磁盘I/O性能来提高producer的响应时间。这也是为什么通常优先把log segment功能放在可以快速响应的磁盘中（比如，SSD）。这样使得flush进程把“脏数据”写入磁盘前，“脏页”数目就减少了，可以设置 vm.dirty_background_ratio（表示占用系统内存的百分比）参数的值为 10 以下。大部分应用场景下，vm.dirty_background_ratio 设置为 5 就够用了，要注意了：这个参数值不能设置为 0 ，因为设置为 0 后会引起内核持续刷“脏页”，使得内核的buffer write功能没法施展。
“脏页”的总量可以通过 vm.dirty_ratio 来改变，默认值是 20 （此处也是百分比），这个值的设置范围较大，一般建议设置 60 到 80 为合理的值。但是 vm.dirty_ratio 参数也引来了不小的风险，会造成大量unflush的数据在硬刷到磁盘时产生较长的I/O停顿。如果vm.dirty_ratio 值设置的较大时，强烈建议Kafka开启备份功能，以备系统崩溃。
在设置了这些参数后，需要监控Kafka集群运行时“脏页”的数量，当前“脏页”数量可由如下方式查看（/proc/vmstat文件）：
# cat /proc/vmstat | egrep "dirty|writeback" nr_dirty 3875nr_writeback 29nr_writeback_temp 0 |
磁盘
除了考虑磁盘硬件本身和RAID配置外，磁盘的filesystem对Kafka集群的影响最大。虽然有许多filesystem，但最常用的是EXT4或者XFS。在这里XFS文件系统比EXT4稍好，具体原因Google下。
另外一点是，建议开启mount的noatime mount选项。文件系统在文件被访问、创建、修改等的时候会记录文件的一些时间戳，比如：文件创建时间（ctime）、最近一次修改时间（mtime）和最近一次访问时间（atime）。默认情况下，atime的更新会有一次读操作，这会产生大量的磁盘读写，然而atime对Kafka完全没用。
网络
Linux发布版本的网络参数对高网络流量不适用。对于Kafka集群，推荐更改每个socket发送和接收buffer的最大内存：net.core.wmem_default 和 net.core.rmem_default 为128 kb，net.core.wmem_max 和 net.core.rmem_max 为 2 Mb。另外一个socket参数是TCP socket的发送和接收buffer： net.ipv4.tcp_wmem 和 net.ipv4.tcp_rmem。