Kafka 投递语义与可靠性模式：重试、幂等消费与死信队列

生产端：
  幂等生产者 + acks=all → 不丢、单实例重试不重复（at-least-once）

存储端：
  3 副本 + ISR → 已提交消息不丢（持久化）

消费端：
  手动提交偏移量 → 处理成功后提交 = at-least-once
  处理前提交（auto.commit） → 可能丢 = at-most-once
  事务 + 幂等表 → exactly-once

非关键/可容忍丢失：at-most-once（省成本）
默认选择：at-least-once + 幂等消费（性价比最高）
强一致关键业务：exactly-once（事务，见 kafka-transactions 篇）

① 生产端重试：网络抖动 → 同一消息 broker 收两次（幂等生产者可解）
② 消费端提交失败：处理完但 offset 没提交 → 重启重读已处理消息
③ 事务回滚重放：事务失败重来 → 消费者看到两次
④ 分区重平衡：Consumer 被踢出再入组 → 重新消费部分消息

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        process(r);              // 处理业务（如写库）
    }
    consumer.commitSync();       // 全部成功后再提交偏移量
    // 若 process 抛异常：不提交 → 重启后重读该批 → 重复处理
}

-- 消息去重表
CREATE TABLE processed_events (
  msg_id       VARCHAR(64) PRIMARY KEY,   -- 消息唯一 ID（业务幂等键）
  topic        VARCHAR(64) NOT NULL,
  partition    INT        NOT NULL,
  offset       BIGINT     NOT NULL,
  processed_at TIMESTAMP  DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY uk_po (topic, partition, offset)   -- 可选：按位置去重
);

// 消费端：先插幂等表再处理业务
public void process(ConsumerRecord<String,String> r) {
    try {
        dao.insertProcessed(r.key());       // ① 插入幂等表（唯一键）
        applyBusinessLogic(r);              // ② 业务逻辑（同一事务）
        consumer.commitSync();              // ③ 提交
    } catch (DuplicateKeyException e) {
        consumer.commitSync();              // 已处理过 → 跳过并提交
    }
}

// 用 SETNX 加唯一键，TTL 覆盖重试窗口
boolean first = jedis.setnx("event:" + msgId, "1");
jedis.expire("event:" + msgId, 3600);   // 1 小时窗口
if (!first) { /* 已处理，跳过 */ }

① 客户端自动重试：broker 侧超时/可重试错误（生产端 retries）
② 消费端代码重试：单消息处理失败 → 重新 poll 重试
③ 重试 Topic 模式：失败消息投递到「重试队列」，延迟后回读

int attempts = 0;
long baseDelay = 1000;         // 1s 起步
while (attempts < 5) {
    try {
        process(msg);
        break;
    } catch (RetryableException e) {
        attempts++;
        long delay = baseDelay * (1L << (attempts - 1));  // 1,2,4,8,16s
        delay += new Random().nextInt(500);               // 加抖动防惊群
        Thread.sleep(delay);
    }
}
// 5 次仍失败 → 投递 DLQ

orders 主 Topic → 失败 → orders.retry（延迟 30s 后回读）
    重试 3 次仍失败 → orders.dlq（死信）

命名规范：<topic>.dlq 或 <topic>.<group>.dlq
消息内容：保留原始 payload + header（原因、重试次数、原始 offset）
保留策略：长保留（审计用），别短于业务回溯周期
监控：DLQ 深度/Lag 是核心告警

// 消费端：重试耗尽后投递 DLQ
catch (Exception e) {
    if (attempts >= MAX_ATTEMPTS) {
        Map<String, String> headers = Map.of(
            "err-reason", e.getMessage(),
            "err-retries", String.valueOf(attempts),
            "err-original", r.topic() + ":" + r.partition() + ":" + r.offset()
        );
        dlqProducer.send(new ProducerRecord<>("orders.dlq",
            r.key(), r.value(), headers));
        consumer.commitSync();   // 摘除毒消息，继续消费
    }
}

① 落 DLQ（带原因/重试次数）
② 告警（DLQ 深度 > 阈值）
③ 人工/工具排查原因（反序列化错？下游 bug？）
④ 修复后重新投递回主 Topic（重放）
⑤ 定期清理过期 DLQ 消息

生产端：幂等生产者 + acks=all + min.insync=2   → 不丢不重（单实例）
存储端：3 副本 + ISR                            → 持久可靠
消费端：手动提交 + 幂等表 + 重试退避 + DLQ       → 重复无害 + 毒消息隔离

生产端：发送失败率、重试率、delivery.timeout
消费端：消费 Lag、处理失败率、重试次数分布
DLQ：深度、增长速率、重放成功率