🔥 Kafka 消息核心问题详解

🛡️

一、消息可靠性（Exactly-Once Semantics）最核心

核心问题：Kafka 如何保证消息不丢、不重，即 精确一次语义（EOS）？这是生产环境最关键的问题。

1.1 可靠性的三个层次

1.2 生产端保障（Producer → Broker）

# 关键配置
acks = all            # (或 -1) ISR 中所有副本都确认才算成功
retries = Integer.MAX_VALUE   # 无限重试
max.in.flight.requests.per.connection = 1  # 顺序发送
enable.idempotence = true     # 开启幂等性（Kafka 0.11+）

• acks=0：发完就不管了，最快但最不可靠
• acks=1：Leader 写入成功就返回，Leader 挂了可能丢数据
• acks=all/-1：ISR 所有副本都同步完才返回，最可靠但有延迟

1.3 Broker 端保障（副本机制）

# Broker 配置
replication.factor = 3        # 分区副本数 ≥ 3
min.insync.replicas = 2       # 最少同步副本数（必须 > 1）
unclean.leader.election.enable = false  # 不允许非 ISR 节点当选 Leader

ISR（In-Sync Replicas）是跟 Leader 保持同步的副本集合。min.insync.replicas = 2 保证至少 2 个副本写成功，这样即使挂 1 个也不影响。

1.4 消费端保障（Consumer 处理）

# Consumer 最佳实践
enable.auto.commit = false     # 关闭自动提交！手动控制
auto.offset.reset = earliest  # 新消费者从头开始

# 正确的消费流程：
# 1. 拉取消息
# 2. 业务处理（写入 DB / 调用下游）
# 3. 处理成功 → 手动 commit offset
# 4. 处理失败 → 重试 或 记录死信队列

1.5 Kafka 事务 API（跨分区 Exactly-Once）

Kafka 0.11+ 支持跨分区的事务写入，配合幂等生产者实现真正的 EOS：

// Producer 初始化事务
props.put("transactional.id", "my-tx-id");
producer.initTransactions();

try {
    producer.beginTransaction();
    // 向多个 topic/partition 发送消息
    producer.send(new ProducerRecord<>("topic-a", key, value));
    producer.send(new ProducerRecord<>("topic-b", key, value));
    producer.commitTransaction();  // 全部原子提交
} catch (Exception e) {
    producer.abortTransaction();   // 全部回滚
}

🔢

二、消息顺序性高频面试题

核心问题：Kafka 如何保证消息的消费顺序？为什么默认情况下消息可能是乱序的？

2.1 为什么会乱序？

2.2 根本原因图解

Producer ──┬──> Partition 0: msg1 → msg3 → msg5 → msg7  (各自有序)
           ├──> Partition 1: msg2 → msg4 → msg6 → msg8  
           └──> Partition 2: ...

Consumer Group
  ├── C0 消费 P0: msg1, msg3, msg5...
  ├── C1 消费 P1: msg2, msg4, msg6...     ← 各自并行，无法保证全局顺序！
  └── C2 消费 P2: ...

期望输出: msg1, msg2, msg3, msg4 ...
实际可能: msg1, msg3, msg2, msg5 ...      ← ❌ 乱序！

2.3 解决方案

// 方案二示例：按订单 ID 分区，保证同一订单的消息有序
ProducerRecord<String, String> record = 
    new ProducerRecord<>("orders", orderId.toString(), orderJson);
// 相同 orderId → 相同 Partition → 有序 ✓

🔄

三、消息重复消费必问

核心问题：什么情况下会产生重复消息？如何做到幂等消费？

3.1 产生重复的三大场景

3.2 核心解决方案：幂等性（Idempotency）

定义：对同一操作执行一次和执行 N 次，结果相同。这是解决重复的终极武器。

3.3 幂等的常见实现方式

-- 方案：数据库唯一约束
CREATE TABLE orders (
    order_id VARCHAR(64) PRIMARY KEY,  -- 消息中的业务ID作为唯一键
    amount DECIMAL(10,2),
    status VARCHAR(20),
    create_time DATETIME,
    UNIQUE KEY uk_msg_id (message_id)    -- 或者用消息ID做唯一键
);

-- 消费端伪代码
void consume(OrderMessage msg) {
    try {
        db.insert("INSERT IGNORE INTO orders VALUES (?, ?, ?, ?, ?)",
                  msg.orderId, msg.amount, msg.status, now(), msg.messageId);
        // INSERT IGNORE：如果存在则跳过（幂等！）
        commitOffset();
    } catch (Exception e) {
        sendToDLQ(msg);  // 死信队列
    }
}

📦

四、消息积压（Backlog）运维高频

核心问题：Consumer 消费速度跟不上 Producer 发送速度，导致消息大量堆积怎么办？

4.1 积压的原因分析

4.2 应急处理步骤（⭐ 重点掌握）

Step 1: 发现积压（监控告警）
├── Lag 监控：consumer lag > 阈值告警
├── 磁盘监控：Broker 磁盘使用率 > 80%
└── 日志监控：日志出现 "lag increasing"

Step 2: 紧急扩容（临时方案）
├── 将当前 Consumer 停掉（保留 Offset）
├── 新建一个 Topic（partitions = 原来的 N 倍，比如 10 倍）
├── 编写临时消费程序：
│   ├── 只做转发：从原 Topic 快速读 → 批量写到新 Topic
│   ├── 不做任何业务处理！（最大化速度）
│   └── 可以启动 M 个 Consumer 同时跑
├── 启动 N 倍于原来的正式 Consumer 消费新 Topic
└── 积压消化完后，恢复原有架构

Step 3: 根因修复（长期方案）
├── 优化 Consumer 处理逻辑（异步、批处理）
├── 下游扩容（DB 连接池、缓存加速）
└── 上游限流（Producer 速率控制）

4.3 积压处理的架构图

正常架构：
Producer → [Topic-A] → Consumer(3个) → 业务DB

积压应急架构：
Producer → [Topic-A] → 临时转发程序(10个) → [Topic-B(10x分区)] → Consumer(30个) → 业务DB
                      ↑ 只转发不处理                    ↑ 扩容消费              ↑ 恢复正常速度

4.4 预防措施

1. Lag 监控：设置 consumer-lag 告警阈值，超过即报警
2. 容量规划：提前评估峰值 QPS，预留 2-3 倍余量
3. 降级策略：积压严重时丢弃非关键消息或降级处理
4. 批量消费：max.poll.records 适当调大，减少 RPC 次数
5. 异步消费：Consumer 收到消息后放入内存队列，后台线程池异步处理

❌

五、消息丢失P0 级事故

核心问题：消息在哪些环节可能丢失？如何全方位防止消息丢失？

5.1 消息传递全链路

每个环节都可能丢消息 ↓

5.2 各环节丢失场景与对策

5.3 防丢失完整配置清单

# ========== Producer ==========
bootstrap.servers = broker1:9092,broker2:9092,broker3:9092
acks = all                              # 必须所有 ISR 副本确认
retries = 2147483647                    # 无限重试
max.in.flight.requests.per.connection = 5 # 配合 enable.idempotence 使用
enable.idempotence = true               # 开启幂等（防重复+防乱序）

# delivery.timeout.ms = 120000          # 总超时时间
# retry.backoff.ms = 100                # 重试间隔

# ========== Broker（server.properties）==========
# --- 副本相关 ---
default.replication.factor = 3          # 默认副本数 = 3
min.insync.replicas = 2                 # 至少 2 个副本同步
unclean.leader.election.enable = false  # 禁止脏选举

# --- 刷盘相关 ---
# log.flush.interval.messages = 10000   # 每 1 万条刷盘（按需调整）
# log.flush.interval.ms = 1000          # 每 1 秒刷盘（按需调整）
# 注意：默认依赖 OS 异步刷盘，性能更好；强一致性需开启主动刷盘

# --- ISR 相关 ---
replica.lag.time.max.ms = 30000         # Follower 超过 30s 不同步则踢出 ISR
num.recovery.threads.per.data.dir = 3

# ========== Consumer ==========
enable.auto.commit = false              # 【关键】关闭自动提交
auto.offset.reset = earliest            # 新 consumer 从头开始
# max.poll.records = 500                # 每次最多拉取 500 条
# 手动控制：poll() → process() → 成功则 commitSync()