1、消息丢失

• 首先是 「已提交」的消息：当 Kafka 中 N 个 Broker 成功的收到一条消息并写入到日志文件后，它们会告诉 Producer 端这条消息已成功提交了，那么这时该消息在 Kafka 中就变成 "已提交消息" 了。这里的 N 个 Broker 主要取决于对 "已提交" 的定义， 这里可以选择只要一个 Broker 成功保存该消息就算已提交，也可以是所有 Broker 都成功保存该消息才算是已提交。

• 其次是 「最大限度的持久化保证不丢失」，也就是说 Kafka 并不能保证在任何情况下都能做到数据不丢失。即 Kafka 不丢失数据是有前提条件的。假如这时你的消息保存在 N 个 Broker 上，那么前提条件就是这 N 个 Broker 中至少有1个是存活的，就可以保证你的消息不丢失。

1. Producer 端发送消息给 Kafka Broker 端。

2. Kafka Broker 将消息进行同步并持久化数据。

3. Consumer 端从 Kafka Broker 将消息拉取并进行消费。

1.1 Producer端避免消息丢失

• 网络原因：由于网络抖动导致数据根本就没发送到 Broker 端。

• 数据原因：消息体太大超出 Broker 承受范围而导致 Broker 拒收消息。

1. 设置 acks=all。这是保证消息不丢失的最严格设置，表示消息发送到 Kafka 集群后要等分区的 ISR 集合中最小副本数量（由 min.insync.replicas 参数控制）个副本同步成功后，这条消息才算发送成功。简单来说，acks = all + min.insync.replicas = N 配置表示消息发送后要等该分区的 ISR 集合中最少 N 个副本同步成功后，这条消息才算发送成功。

2. 设置重试：
1. 设置重试次数 retries 大于 0。该参数表示 Producer 端发送消息的重试次数，在 Kafka 2.4 版本中默认设置为 Integer.MAX_VALUE ，说明会一直进行重试直到 Broker返回 Ack。
2. 设置重试时间间隔：retry.backoff.ms 该参数表示消息发送超时后两次重试之间的间隔时间，默认值为 100ms。为了避免无效的频繁重试，推荐设置为 300ms。

3. 发送消息使用带回调函数的方法：使用 Producer.send(msg, callback) 发送消息。这样一旦发现发送失败， 就可以做针对性处理。不要使用 Producer.send(msg) 发送消息，该方法会立即返回，由于没有回调，可能因网络原因导致 Broker 并没有收到消息，此时就丢失了。

1.2 Broker端避免消息丢失

1. 设置多副本：replication.factor参数表示分区副本的个数。建议设置 replication.factor >= 3，这样如果 Leader 副本异常 Crash 掉，Follower 副本会被选举为新的 Leader 副本继续提供服务。

2. 设置 ISR 最小同步副本数：min.insync.replicas参数表示 ISR 列表最少需要有多少个副本同步消息成功，该消息才算写入成功。这是配合 acks=all 和 replication.factor 的关键参数，建议设置 min.insync.replicas > 1，这样才可以提升消息持久性，保证数据不丢失。推荐设置成：replication.factor = min.insync.replicas + 1 ，最大限度保证系统可用性。
• 例如：replication.factor=3, min.insync.replicas=2, acks=all。

3. 禁用非 ISR 列表的 Follower 竞选 Leader：unclean.leader.election.enable参数表示是否允许非 ISR 列表的 Follower 被选举为 Leader ，默认是 true。如果一个 Follower 不在 ISR 列表里面，通常是因为它的数据落后 Leader 太多，那么一旦它被选举为新的 Leader， 数据就会丢失，因此我们要将其设置为 false，防止此类情况发生。

1.3 Consumer端避免消息丢失

1. 获取元数据并从 Kafka Broker 集群拉取数据。

2. 处理消息，并标记消息已经被消费，提交 Offset 记录。

1. 如果采用自动提交 Offset 的方式。在提交完 Offset 之后 Conumer 宕机了，在 Conumer 重启之后已经开始拉取下一条消息了，那么上一条消息就算丢了。

2. 如果采用手动提交 Offset 的方式，这里有两种情况：
1. 「先提交 Offset，后处理消息」，如果此时处理消息的时候异常宕机，由于 Offset 已经提交了，待 Consumer 重启后，会从之前已提交的 Offset 下一个位置重新开始消费， 之前未处理完成的消息不会被再次处理，对于该 Consumer 来说消息就丢失了。
2. 「先处理消息，在进行提交 Offset」， 如果此时在提交之前发生异常宕机，由于没有提交成功 Offset， 待下次 Consumer 重启后还会从上次的 Offset 重新拉取消息，不会出现消息丢失的情况， 但是会出现重复消费的情况，这里只能业务自己保证幂等性。

1. 禁用自动提交位移，采用手动提交位移。设置 enable.auto.commit = false 。而且必须要先处理消息，处理成功之后再进行提交 Offset。

2. 选择合适的提交方式：
• 同步提交 (commitSync())：更可靠，会阻塞直到提交成功或遇到错误。
• 异步提交 (commitAsync())：性能更好，但可能提交失败时不会自动重试（通常可以配合回调函数进行重试）。
• 如果是手动提交，我们需要将 commitSync 和 commitAsync 组合使用才能达到最理想的效果。第一次使用异步提交，如果发生异常再使用同步提交，这样既可以保证发送的效率，又能规避因网络抖动引发的瞬时错误。

3. 业务保证幂等性。消息重复消费的情况无法避免，业务必须自己进行幂等性控制（如检查数据库唯一键）。

2、消息重复

• Producer 端重复发送

• Consumer 端重复消费

2.1 Producer端避免消息重复

1. 生产者发送消息后，Kafka Broker 需要返回一个确认（Ack）。如果 Broker 因网络波动、重启等原因，未能及时返回 Ack，生产者会认为消息发送失败。

2. 发送消息如果配置了重试机制，为保证消息不丢失生产者就会重新发送消息。

3. 如果第一次发送实际上已经成功，只是 Ack 确认丢失，那么重试就会导致同一条消息被 Broker 接收两次。

• Producer 端启用幂等性（Idempotent Producer）：在生产者配置中设置 enable.idempotence = true 开启幂等性，可以实现消息自动去重，缺点是只能保证单会话和单分区的幂等性。

• Producer 端启用事务：保证生产者发送到多个分区的消息要么全部写入成功，要么全部写入失败。适用于需要保证多个分区的多条消息的原子性场景。

2.2 Consumer端避免消息重复

1. 消费者处理完消息后，需要提交偏移量（Offset）。如果消费者处理完消息，但在提交 Offset 之前突然崩溃（如进程被杀、宕机）。

2. 当消费者重启后还会从上次的 Offset 重新拉取消息，出现重复消费的情况。

• 使用消息去重表（基于数据库的唯一索引）：单独建立一张“消息处理表”，将消息的唯一标识（如订单ID、业务ID）作为数据库表的主键或唯一索引。把消息插入和业务处理放在一个事务里面，先把消息插入该表，如果插入成功则处理业务逻辑；如果插入失败说明该消息已经被处理过，则跳过。

• 基于 Redis 的分布式锁：把消息的唯一标识作为 key，在消费消息时使用 Redis 对该 key 加锁，在 Offset 提交完成后，再在 Redis 中删除该标识的 key。

• 基于数据库的乐观锁：给数据增加一个版本号字段（version），每次更数据前，比较当前数据的版本号是否和消息中的版本号一致，如果不一致就拒绝更新数据，更新数据的同时将版本号 +1。

3、消息积压

3.1 如何发现消息积压

1. Kafka 自带工具：kafka-consumer-groups.sh：这是最直接的工具。

• 输出中的 CURRENT-OFFSET 是消费者当前消费到的位置。

• LOG-END-OFFSET 是分区中最新消息的位置。

• LAG = LOG-END-OFFSET - CURRENT-OFFSET。

1. 监控系统（Grafana + Prometheus + JMX Exporter）：这是生产环境的必备方案。通过 JMX Exporter 将 Kafka Consumer 的 JMX 指标（特别是 kafka.consumer:type=consumer-fetch-manager-metrics,client-id={client-id} 下的 records-lag-max 和 records-lag）暴露给 Prometheus，然后在 Grafana 中绘制漂亮的监控大盘。关键监控指标：
• records-lag-max： 所有分区中最大的积压数（最敏感的指标）。
• records-lag： 每个分区的具体积压数。
• consumption-rate： 消费速率。
• fetch-rate： 向 Broker 拉取消息的速率。

3.2 消息挤压的原因

1. 消费者性能问题（最常见）
• 单条处理逻辑耗时过长： 例如，在消息处理中执行了复杂的业务计算、同步调用外部API（且对方响应慢）、进行了沉重的数据库操作（全表扫描、未加索引）等。
• 分区数/消费者实例过少：分区数是 Kafka 并行度调优的最小单元，如果分区数设置的太少，或者消费者实例太少，会影响消费者组整体消费的吞吐量。
• 低效的消费代码：
• 没有利用多线程或异步处理。对于 I/O 密集型的操作，单线程消费是性能杀手。
• 错误配置，如 max.poll.records 设置得过大，一次拉取太多消息，导致单次 poll 的处理时间过长，甚至可能超过 max.poll.interval.ms，导致消费者被踢出组（rebalance）。
• 没有启用批量处理，而是一条一条处理。
• 消费者故障： 消费者实例宕机或 GC 停顿时间过长，导致消费能力下降或暂时停止。
• 频繁的 Rebalance： 消费者组因为网络问题、心跳超时、处理时间过长等原因发生频繁的再平衡，在 rebalance 期间消费者会停止消费。

2. 生产者流量激增
• 业务高峰（如秒杀、促销活动）、大数据导入任务等，导致生产消息的速率远远超过了消费者的正常处理能力。

3. 资源瓶颈
• CPU/内存： 消费者应用所在的机器资源被其他进程占用，或本身分配不足。
• I/O： 消费者依赖的数据库、缓存、外部存储等成为瓶颈，响应变慢，拖累了消费速度。
• 网络： 消费者与 Kafka Broker 之间，或消费者与下游系统（如DB）之间的网络带宽不足或延迟过高。

3.3 解决挤压问题

1. 紧急扩容：最快最直接的临时方案，适用于消费者消费能力不足的场景。
1. 增加消费者实例： 这是最有效的方法。但注意，一个分区只能被同一个消费者组内的一个消费者消费。因此，消费者实例数不能超过主题的分区总数，否则多出来的消费者是空闲的。
2. 增加消费者的硬件资源： 临时给消费者Pod/容器分配更多的 CPU 和内存。

2. 新增Topic，快速转发：如果消息量远超消费者的消费能力（例如挤压的消息数是平时的上百倍），单纯增加消费者实例的消费速度还是太慢了，可以考虑这种方案。
1. 新增 Topic，分区数是原来 Topic 的多倍。
2. 新增临时消费者程序，不做业务处理，只负责转发消息，让其将收到的消息快速转发到新增的 Topic。
3. 启动多个消费者（是原来的多倍），消费新 Topic 的分区，并处理业务逻辑。
4. 在业务低峰期，增加原来的 Topic 的分区数量和消费者数量（会触发 Rebalance）

3. 优化消息的处理能力：提高单条消息的处理能力
1. 多线程处理： 使用线程池，将拉取到的消息分批交给多个线程并行处理。
2. 避免耗时调用：避免在消费循环中进行同步的远程调用，考虑异步化或批量调用。

4. 生产者背压：虽然 Broker 不能直接告诉生产者“慢点发”，但生产者可以通过配置感知到 Broker 的压力。
1. acks=all： 生产者需要等待所有 ISR（同步副本）确认。如果 Broker 或 Follower 副本压力大，响应变慢，会直接降低生产者的发送速率。这是生产端最有效的“自我限流”机制，也是一种强烈的背压体现。
2. max.in.flight.requests.per.connection： 定义了每个连接上最多有多少个未收到响应的请求（默认值为 5）。如果这个值被占满，生产者端的发送线程会被阻塞。

4、顺序消息

• 单分区方案：可以保证发送到该 Topic 的所有消息的顺序性。但是完全丧失了 Kafka 的并行处理和水平扩展能力，吞吐量极低，不推荐使用。
• 创建只有一个 Partition 的 Topic
• 全局使用一个 Producer
• 全局使用一个 Consumer（并严格到一个消费线程）

• 使用消息 Key（推荐）：生产者发送消息时，可以指定一个 Key。Kafka 通过 Key 的哈希值来决定消息应该被发送到哪个分区。所有具有相同 Key 的消息都会被发送到同一个分区。
• 在生产中，例如以订单 ID 作为消息的 Key。这样，所有相同订单 ID 的消息都会被路由到同一个分区。

• 指定 partition：生产者在发送消息构建 ProducerRecord 时，可以直接传入一个 partition 参数，指定 partition 的值，将需要顺序处理的所有消息发送到同一个分区。
• 在生产中，可以为每个用户指定 partition（例如用户 ID 对分区数取余，得到的值就是 partition），这样可以保证该用户的所有消息都发送到同一个分区。这样既可以为 Topic 设置多个分区提高吞吐量，也可以保证单个用户的消息顺序性。

• max.in.flight.requests.per.connection。这个参数的作用是在发送阻塞前对于每个连接，正在发送但是发送状态未知的最大消息数量。如果设置大于 1，那么就有可能存在有发送失败的情况下，因为重试发送导致的消息乱序问题。所以我们应该将其设置为 1，保证在后一条消息发送前，前一条的消息状态已经是可知的。

5、Kafka为什么这么快

1. 分区设计：每个 Topic 分成多个 Partition，分布在不同的 Broker 上，可以实现生产者向多个 Partition并行发送消息，消费者从多个 Partition并行读取消息，并且还增加 Partition 和 Broker 提升整体吞吐量。

2. 磁盘顺序读写：这是最重要的原因，每个 Partition 是一个独立的日志文件。Kafka 将 message 通过顺序追加写的方式追加到 Partition 本地磁盘文件末尾，这种基于磁盘的顺序读写性能很高。

3. 页缓存（Page Cache） + 零拷贝（Zero-Copy）
• Page Cache 技术：Kafka 大量利用操作系统内核的 Page Cache，而不是 JVM 堆内存。这样对内存的使用效率更高，避免了在 JVM 中创建大量对象带来的 GC 压力和内存开销。
• 零拷贝（Zero-Copy）技术：零拷贝技术直接调用mmap 或者sendfile 内核函数，数据无需从内核空间拷贝到用户空间（JVM），减少了 2-3 次上下文切换和内存拷贝带来的性能开销，显著提升吞吐量。
• Producer—>Broker：mmap。Producer 生产的数据存到 Broker，写入 Broker 上内核 mmap 技术映射的 Page Cache 区域，在内核空间就可以完成落盘，数据没有经过用户空间，消除了内核空间和用户空间之间的拷贝开销（2次）。
• Broker—>Consumer：sendfile。Broker 上的消息被 Consumer 消费的时候，磁盘数据通过 DMA 拷贝到内核 Page Cache 后，使用 sendfile 直接通过 DMA 拷贝到网卡缓冲区（NIC Buffer），数据没有经过用户空间，消除了内核空间和用户空间之间的拷贝开销（2次）。

4. 批量处理：无论是 Producer 还是 Consumer，Kafka 都极力鼓励批量操作。
• 批量发送：生产者发送消息不是单条发送的，而是积累多条消息后使用批次（Batch）批量发送，可以减少网络 IO 次数。可以通过 linger.ms 和 batch.size 参数控制。
• 压缩：Kafka 在批次消息发送前，对批次消息进行压缩（如 gzip、snappy、lz4、zstd），减少网络传输的文件大小。消费者在消费压缩消息时会自动解压缩。
• 批量拉取：消费者按批次拉取消息，降低网络往返开销。

6、Kafka分区设计的作用

• 横向扩展：通过将一个 Topic 划分为多个分区，这些分区分散到 Kafka 集群的多个 Broker上，实现了数据的分布式存储和负载均衡。

• 并行消费：消费者以 Consumer Group 为单位消费 Topic。Topic 下的每个分区在同一时间只能被同一个消费者组内的一个消费者消费，Topic 的多个分区可以分配给多个消费者分配，这样提高了消息消费的吞吐量。

7、Kafka的数据倾斜问题

• 分区级倾斜：这是最常见的数据倾斜类型，发生在单个主题（Topic）内部。
• 问题现象：某个主题下的消息绝大部分都被写入到少数几个分区中，而其他分区则很少有数据。
• 原因：
• 消息设置热门 key：如果消息指定了 key，分区器会把相同 key 值的消息发送到同一个分区。如果某些 key 的出现频率远高于其他 key（例如，日志中的某个热门用户ID、某个异常状态码），那么这些“热点 key”总是会被路由到同一个特定分区，导致该分区数据量激增。
• 自定义分区策略不合理：如果业务自定义了分区器（Partitioner），但逻辑有缺陷，也可能导致分布不均。
• 解决方案：
• 设计合理的 key，避免使用热点 Key：
• 如果业务允许，最佳实践是将消息的 key 设置为 null，让生产者使用轮询策略，实现绝对均匀。
• 如果必须要有 key（例如需要保证同一 key 的消息有序性），可以尝试对原始 key 进行加工，例如在 key 后添加随机数后缀（user123_<random>），然后再进行哈希。这样既能在一定程度上保证相同原始 key 的消息有序（需要消费者端做额外处理），又能将数据打散。
• 选择一个基数（Cardinality）大、分布均匀的字段作为 key。例如，使用 用户ID 比使用 用户性别 作为 key 要好得多，因为前者的可能值更多，分布更随机。
• 自定义合理的分区器：自定义分区器。可以根据消息的某些属性，结合当前的分区负载情况，实现一个更合理的路由逻辑。

• 主题/ Broker 级倾斜：这种倾斜发生在 整个 Kafka 集群层面。
• 问题现象：少数几个 Broker 节点承载了绝大部分的分区副本（Leader 和 Follower），导致这些 Broker 的磁盘 I/O、网络流量和 CPU 负载远高于其他节点。
• 原因：
• 自动分区分配策略：当创建新主题或进行分区重分配时，Kafka 的默认分配算法（如 default.replication.factor）可能没有充分考虑集群的现有负载，导致新分区被集中创建在少数 Broker 上。
• 手动分配不当：手动指定分区副本位置时，没有均匀分布。
• 解决方案：
• 使用工具重新分配分区：Kafka 提供了 kafka-reassign-partitions.sh 工具，可以生成一个计划，将分区副本从高负载 Broker 迁移到低负载 Broker 上，从而实现集群级别的负载均衡。
• 增加分区数：对于已经存在倾斜的主题，可以增加其分区数量（kafka-topics.sh --alter）。注意：这只会对新数据生效。新的消息（尤其是 key 为 null 的）会分布到新的分区中，但旧数据仍然留在原来的分区里。这是一种“稀释”解决方案，无法解决历史数据的倾斜问题。增加分区后通常也需要相应地增加消费者实例数。

8、Rebalance问题