Kafka 消息不丢不重：生产端到消费端的完整配置

消息队列的可靠性，绕不开三个问题：会不会丢、会不会重、顺序能不能保证。这三个问题的答案互相牵制——想不丢就要重试，重试就会重复，想严格顺序就只能单分区单线程。

这篇文章按「生产端 → Broker → 消费端」三段拆解，每段给出会丢消息的具体环节和对应配置，最后是「至少一次」语义下的幂等方案。

一、先看清消息在哪些环节可能丢

Producer ──①发出──> Broker（Leader ──②写盘──> 副本同步）──③──> Consumer ──④处理──> 业务库

四个丢失点：

① 和 ③ 是最常见的两个坑，而且都是默认行为导致的。

二、生产端：acks 与重试

acks 的三个取值

# acks=0：发出去就认为成功，不等任何确认
# 优点：吞吐最高（约 100 万条/秒）
# 缺点：Broker 挂了消息就没了，且生产端不知道
acks=0

# acks=1：leader 写盘后返回确认（默认）
# 优点：延迟低，吞吐高（约 30 万条/秒）
# 缺点：leader 写盘后、副本同步前宕机 → 消息丢失
acks=1

# acks=all（-1）：leader + 所有 ISR 副本都写盘后返回
# 优点：最强保证
# 缺点：延迟高（多一次网络往返），吞吐约 10 万条/秒
acks=all

关键细节：acks=all 只保证「所有 ISR（In-Sync Replicas）都写了」。如果副本落后太多被踢出 ISR，acks=all 就退化成 acks=1。要避免这个坑，必须同时配：

# 最小 ISR 数量，少于这个数就拒绝写入（而不是降级）
min.insync.replicas=2

配合 replication.factor=3，含义是：3 个副本里至少 2 个同步成功才算写入成功。如果只剩 1 个副本活着，生产端会收到 NotEnoughReplicasException，而不是静默降级。

这三个参数的组合关系必须一起看：

replication.factor = 3
min.insync.replicas = 2
acks = all

→ 可以容忍 1 个 Broker 宕机而不丢消息
→ 2 个 Broker 同时宕机时，写入失败（但不会丢）

只设 acks=all 而不设 min.insync.replicas，等于给自己一个虚假的安全感。

retries 与 delivery.timeout.ms

# 重试次数（默认 Integer.MAX_VALUE，配合超时控制）
retries=2147483647

# 单条消息的总超时（包含重试时间），默认 120000
delivery.timeout.ms=120000

# 两次重试的间隔基数
retry.backoff.ms=100

Kafka 2.1+ 之后，retries 的默认值已经是 MAX_VALUE，真正的限制是 delivery.timeout.ms。这个设计更合理——按时间而不是次数限制，避免慢网络下无限重试。

但重试会引入「重复」：生产端发送成功但没收到 ack（网络抖动），重试后消息重复。所以 Kafka 2.1 引入了幂等生产者。

幂等生产者

enable.idempotence=true

开启后 Kafka 为每个生产者分配 PID，并为每条消息附带递增的序列号。Broker 会去重：

PID=1000, seq=5   → 已收到，丢弃
PID=1000, seq=6   → 新消息，写入

enable.idempotence=true 会自动把 acks 设为 all、retries 设为 MAX_VALUE、max.in.flight.requests.per.connection 限到 5。它是安全默认值，应该总是开启。

限制：幂等只在单个生产者会话内有效。生产者重启后 PID 变了，去重失效。要跨会话保证，需要事务。

事务

props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-order-1");

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", orderId, orderJson));
    producer.send(new ProducerRecord<>("order-events", orderId, eventJson));
    producer.commitTransaction();          // 原子提交
} catch (Exception e) {
    producer.abortTransaction();           // 全回滚
}

事务的价值在于跨 topic 的原子性：要么两条消息都可见，要么都不可见。这是「订单表 + 事件表」这类场景的正确做法。价格上，事务让吞吐下降 20～30%。

一个容易被忽略的问题：异步发送没处理回调

// 差：异常被吞掉
producer.send(record);

// 好：必须处理回调
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("发送失败 topic={} key={}", record.topic(), record.key(), exception);
        // 落库待补偿 / 告警
    }
});

send() 返回 Future，不 get() 也不设回调，发送失败完全静默。这是生产端丢消息的头号原因。

三、Broker 端：副本与刷盘

unclean.leader.election

# 默认 true（允许非 ISR 副本成为 leader）
unclean.leader.election.enable=false

设为 true 时，如果所有 ISR 副本都挂了，Kafka 会从落后的副本里选一个当 leader——那个副本缺的数据就永久丢了。

设为 false 时，这种情况分区不可用（OfflinePartition），但不会丢数据。这是可用性与一致性的取舍。 对「不能丢」的场景必须设 false。

刷盘策略

# 每个分区每秒刷盘次数（默认依赖 OS 的 pagecache 回写）
log.flush.interval.messages=10000
log.flush.interval.ms=1000

Kafka 默认不强制刷盘，靠 OS 的 pagecache 异步回写。这意味着 Broker 进程崩溃不丢（数据在 pagecache 里），但机器断电会丢。

多副本已经能覆盖单机断电场景（其他副本上有数据），所以刷盘策略通常不需要改。只有在 replication.factor=1 的单副本场景才必须配。

保留策略

# 按时间保留（默认 7 天）
log.retention.hours=168
# 按大小保留
log.retention.bytes=1073741824

消息被清理掉不算「丢」——但如果是消费者停了两周后重启，偏移量已经超出保留范围，Kafka 会按 auto.offset.reset 处理：

# earliest：从头开始（可能重复大量消息）
# latest：从最新开始（中间的消息全丢）
auto.offset.reset=earliest    # 或用 none，直接抛异常

生产环境建议 none——让程序明确报错，而不是静默跳过一段数据。

四、消费端：手动提交与幂等

关掉自动提交

# 默认 true，消费者每 5 秒提交一次偏移量
# 问题：消息已提交但业务还没处理完，此时宕机 → 消息丢失
enable.auto.commit=false

正确的手动提交时机

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        try {
            process(r);                      // 业务处理
        } catch (RetryableException e) {
            // 可重试：不能提交偏移量，抛出后重新消费
            throw e;
        } catch (Exception e) {
            // 不可重试：记录到死信队列，然后跳过
            deadLetter.send(r);
            log.error("处理失败，进死信", e);
        }
    }
    consumer.commitSync();                   // 一批处理完再提交
}

关键点：commitSync 的位置在「整批处理完之后」。 如果放在循环内部（每处理一条提交一次），能减少重复但吞吐下降明显（每次提交都是一次网络往返）。

提交与处理的三种组合

结论：没有幂等就没法做到「不丢且不重」。 所以消费端必须解决幂等。

幂等的三种实现

方案 A：数据库唯一约束（最可靠）

CREATE TABLE order_consume_log (
    msg_id      VARCHAR(64) PRIMARY KEY,
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
);

-- 与业务操作在同一个事务里
BEGIN;
INSERT INTO order_consume_log (msg_id) VALUES ('msg-12345');   -- 重复会抛 DuplicateKey
UPDATE inventory SET count = count - 1 WHERE sku_id = 'A';
COMMIT;

主键冲突 → 说明已处理过 → 直接 ack 跳过。这是最可靠的方式，因为幂等判断和业务操作在同一个事务里。

方案 B：状态机

// 订单状态：CREATED → PAID → SHIPPED
// 消费者处理"支付成功"消息时检查当前状态
if (order.getStatus() != CREATED) {
    log.info("订单已是 {} 状态，跳过重复消息", order.getStatus());
    return;
}
order.setStatus(PAID);

只适用于状态单向流转的场景。如果状态可能回退（退款），就不适用。

方案 C：Redis SETNX（快速但有风险）

String key = "consume:" + msgId;
Boolean first = redis.set(key, "1", SetParams.setParams().nx().ex(86400));
if (!first) return;      // 已处理过
process(msg);

风险：Redis 写入成功但业务处理失败 → 消息被标记为「已消费」但实际没处理，永久丢失。所以只能用于「失败了也无所谓」的场景。

方案 A + Redis 前置 是常见组合：Redis 挡住 99% 的重复，DB 唯一约束兜底保证正确性。

五、顺序性

Kafka 只保证单分区内有序。跨分区没有全局顺序。

// 想保证同一订单的消息有序 → 用订单号做 key
new ProducerRecord<>("orders", orderId, payload);

同一个 key 一定落在同一个分区，因此有序。

但注意：生产端重试会破坏顺序。 如果批次 1 失败重试、批次 2 已经成功，顺序就反了。

# 强制单连接单请求，保证顺序（吞吐大幅下降）
max.in.flight.requests.per.connection=1

或者开幂等（enable.idempotence=true），它会自动把 in.flight 限到 5 且保证顺序——这是更优的方案。

消费端保序的前提是单线程消费。 多线程消费同一个分区的消息必然乱序。要提速又保序，用「分区内按 key 二次分流」：

// 按订单号 hash 到固定的工作线程，保证同订单串行、不同订单并行
int slot = Math.floorMod(orderId.hashCode(), 16);
executors[slot].execute(() -> process(record));

六、可靠性配置清单

把上面的结论收成一份配置：

# ---- 生产端 ----
acks=all
enable.idempotence=true
retries=2147483647
delivery.timeout.ms=120000
max.in.flight.requests.per.connection=5
compression.type=lz4                 # 压缩，省带宽

# ---- Broker ----
replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false
auto.create.topics.enable=false      # 禁止自动建 topic（避免默认参数）
default.replication.factor=3

# ---- 消费端 ----
enable.auto.commit=false
auto.offset.reset=none               # 明确报错而不是静默跳过
isolation.level=read_committed       # 只读已提交的事务消息
max.poll.records=500                 # 单批不要太大，避免处理超时
max.poll.interval.ms=300000          # 处理超时时间，超过则触发 rebalance

max.poll.records 和 max.poll.interval.ms 的关系值得单独说：如果一批 500 条要处理 6 分钟，超过 max.poll.interval.ms（默认 5 分钟），消费者会被踢出组，触发 rebalance，然后这批消息被另一个消费者重新消费——无限循环。

现象是「消息一直重复消费，但业务一直没成功」。修法是减小 max.poll.records 或加大 max.poll.interval.ms，让处理时间落在区间内。

七、出问题时的排查顺序

1. 消息真的没了吗？
   kafka-console-consumer --from-beginning --topic orders
   看 Broker 上到底有没有

2. 如果有 → 消费端问题
   - 消费者组 lag：kafka-consumer-groups --describe --group g1
   - 看 LOG-END-OFFSET 和 CURRENT-OFFSET 的差值
   - 差值为 0 但业务没数据 → 检查是否 catch 住了异常当成功

3. 如果没有 → 生产端问题
   - 检查 send() 是否处理了回调
   - 检查 acks / min.insync.replicas
   - 看 Broker 日志有没有 NotEnoughReplicas

4. 有重复 → 幂等没做好
   - 检查消费日志里同一条消息是否处理了多次
   - 补 DB 唯一约束

小结

「不丢不重」是一组联动的配置，单独改任何一项都没用：

一句话总结：Kafka 只能做到「至少一次」，业务幂等才是「恰好一次」的最后一道防线。 没有幂等设计的消费端，配置再严谨也会出事。