Kafka 消费者端幂等设计方案（3 种实现 + 选型决策矩阵）

一、问题定义

当前架构：Spring Boot 2.7 + Kafka 3.x + MySQL 8（ShardingSphere 分库分表）。核心痛点有二：去重表跨分片查询 P99 从 8ms 飙到 45ms；offset commit 与业务写入非原子，崩溃恢复存在约 200ms 重复消费窗口。约束条件：消费 TPS 峰值 5000、P99 延迟 < 50ms、at-least-once 语义不可丢消息、不引入 Redis Cluster 之外的新中间件。

幂等消费的核心命题：在 at-least-once 投递语义下，保证同一条业务消息只被处理一次，且去重检查的延迟不显著增加消费 P99。

下面给出三种方案，从"纯 DB"到"引入 Redis"再到"Kafka 原生能力 + 分层缓存"，逐层递进。每种方案都给出可编译代码、失效边界分析和 ShardingSphere 适配说明。

二、方案一：DB 唯一键去重（纯 DB 方案）

核心思路

利用业务表或独立去重表的唯一索引，将"是否已处理"的判断转化为 INSERT 是否冲突。这是最经典的幂等模式，不依赖任何外部缓存。关键优势：去重判断和业务写入在同一个 DB 事务内完成，天然原子。

核心代码（Java，Spring Boot 2.7）

import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import com.fasterxml.jackson.databind.ObjectMapper;

@Component
public class IdempotentOrderConsumer {

    private final OrderEventRepository eventRepo;
    private final ObjectMapper mapper = new ObjectMapper();

    public IdempotentOrderConsumer(OrderEventRepository eventRepo) {
        this.eventRepo = eventRepo;
    }

    @KafkaListener(topics = "order-events", groupId = "order-svc")
    @Transactional
    public void onMessage(String payload, Acknowledgment ack) {
        OrderEvent event = parse(payload);
        try {
            // 唯一键 = (order_id, event_type)，ShardingSphere 按 order_id 路由
            eventRepo.save(new OrderEventEntity(
                event.getOrderId(),
                event.getEventType(),
                System.currentTimeMillis(),
                payload
            ));
        } catch (DuplicateKeyException e) {
            // 重复消息：安全跳过，不抛异常
            ack.acknowledge();
            return;
        }
        // 业务处理（与去重 INSERT 在同一事务内）
        processOrder(event);
        ack.acknowledge();
    }

    private OrderEvent parse(String json) {
        try { return mapper.readValue(json, OrderEvent.class); }
        catch (Exception e) { throw new IllegalArgumentException(e); }
    }

    private void processOrder(OrderEvent event) {
        // 实际业务逻辑
    }
}

DDL（ShardingSphere 分片键 = order_id）：

CREATE TABLE order_event (
    id          BIGINT AUTO_INCREMENT PRIMARY KEY,
    order_id    VARCHAR(64) NOT NULL,
    event_type  VARCHAR(32) NOT NULL,
    created_at  BIGINT NOT NULL,
    payload     TEXT,
    UNIQUE KEY uk_order_event (order_id, event_type)
) ENGINE=InnoDB;
-- ShardingSphere 配置:
--   actualDataNodes = ds${0..7}.order_event_${0..15}
--   shardingColumn  = order_id
--   algorithmType   = MOD

失效边界分析

| 场景 | 行为 | 风险等级 | |---|---|---| | Broker 宕机 + offset 未提交 | 重启后重新投递，INSERT 冲突 → 跳过 | 安全 | | 消费者 rebalance | 分区重新分配，新消费者从上次 committed offset 开始；若 commit 滞后，窗口内消息重复 → 唯一键兜底 | 安全 | | 网络分区（消费者 ↔ DB） | INSERT 超时 → 事务回滚 → offset 未 commit → 重新消费 | 安全，但 DB 不可用期间消费停滞 | | ShardingSphere 路由错误 | order_id 路由到错误分片 → 同一 orderId 可能落到不同分片 → 唯一键跨分片失效 | 高危：必须确保分片键与唯一键首列一致 | | DB 主从切换 | 写入主库成功但 offset 未 commit，切主后旧主数据丢失 | 极低概率，需确认半同步复制 |

ShardingSphere 适配要点

• 分片键必须与唯一键首列一致（均为 order_id），否则跨分片唯一性无法保证。
• 去重 INSERT 是写操作，走主库，不存在跨分片读的问题——这是相对"先 SELECT 再 INSERT"的巨大优势，直接规避了你当前 P99 45ms 的痛点。
• 若业务表本身已有唯一键，可直接复用，无需独立去重表。

三、方案二：Redis SETNX + TTL 去重（引入 Redis）

核心思路

用 Redis 的 SET key value NX EX ttl 原子操作做去重判断，命中则跳过。Redis 单线程模型天然无锁，P99 延迟通常在 1~3ms，远低于 DB 的 8~45ms。注意：Redis 在这里是加速层，不是唯一保障——业务表仍需保留唯一键兜底。

核心代码（Java，Spring Boot 2.7）

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.Duration;

@Component
public class RedisIdempotentConsumer {

    private static final String KEY_PREFIX = "idem:";
    private static final Duration TTL = Duration.ofMinutes(30);

    private final StringRedisTemplate redis;
    private final OrderService orderService;
    private final ObjectMapper mapper = new ObjectMapper();

    public RedisIdempotentConsumer(StringRedisTemplate redis,
                                    OrderService orderService) {
        this.redis = redis;
        this.orderService = orderService;
    }

    @KafkaListener(topics = "order-events", groupId = "order-svc")
    public void onMessage(String payload, Acknowledgment ack) {
        OrderEvent event = parse(payload);
        String key = KEY_PREFIX + event.getOrderId() + ":" + event.getEventType();
        // SET NX EX 原子操作：返回 null 表示 key 已存在
        Boolean first = redis.opsForValue().setIfAbsent(key, "1", TTL);
        if (Boolean.FALSE.equals(first)) {
            ack.acknowledge(); // 重复，跳过
            return;
        }
        try {
            orderService.process(event);
            ack.acknowledge();
        } catch (Exception e) {
            // 业务失败：删除去重标记，允许重试
            redis.delete(key);
            throw e;
        }
    }

    private OrderEvent parse(String json) {
        try { return mapper.readValue(json, OrderEvent.class); }
        catch (Exception e) { throw new IllegalArgumentException(e); }
    }
}

失效边界分析

| 场景 | 行为 | 风险等级 | |---|---|---| | Redis 主从切换（failover） | 若用主从 + Sentinel，切换瞬间主库未同步的 key 丢失 → 去重标记消失 → 重复处理 | 中等：用 AOF everysec 可将窗口降到约 1s | | Redis 宕机 | setIfAbsent 异常 → 消费者抛异常 → offset 不 commit → 重试 | 安全（fail-safe），但消费停滞 | | TTL 过期 | 30min 后 key 过期，同一消息再次投递 → 去重失效 | 需 TTL > 最大重投窗口（通常 5~10min 足够） | | rebalance | 同方案一，窗口内重复 → Redis 去重兜底 | 安全 | | 网络分区（消费者 ↔ Redis） | setIfAbsent 超时 → 异常 → 重试 | 安全，需配置合理的 command timeout |

ShardingSphere 适配

• Redis 去重完全绕开了分库分表，不存在跨分片问题。
• 但业务写入仍走 ShardingSphere，需确保业务表自身有唯一键兜底——Redis 是加速层，不是唯一保障。
• 推荐"Redis 去重 + DB 唯一键"双保险架构。

四、方案三：Kafka 分区路由 + 内存 LRU + DB 兜底（分层去重）

核心思路

利用 Kafka 的分区顺序性（同一 key 落同一分区 → 单线程顺序消费），在消费者内存中维护一个 LRU 去重窗口。热路径（99%+ 的重复）在内存中拦截（P99 < 0.1ms），冷路径（窗口外的重复，如 rebalance 后的历史消息）落 DB 兜底。不引入任何新中间件。

核心代码（Java，Spring Boot 2.7）

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class TieredIdempotentConsumer {

    private static final int WINDOW_SIZE = 10_000;

    // 每个分区一个 LRU 窗口
    private final Map<String, LruSet> partitionWindows = new ConcurrentHashMap<>();
    private final OrderEventRepository eventRepo;
    private final ObjectMapper mapper = new ObjectMapper();

    public TieredIdempotentConsumer(OrderEventRepository eventRepo) {
        this.eventRepo = eventRepo;
    }

    @KafkaListener(topics = "order-events", groupId = "order-svc",
                   concurrency = "4")
    public void onMessage(ConsumerRecord<String, String> record) {
        String key = record.key(); // 约定: order_id:event_type
        int partition = record.partition();

        // 第一层：内存 LRU（热路径，P99 < 0.1ms）
        LruSet window = partitionWindows.computeIfAbsent(
            String.valueOf(partition), p -> new LruSet(WINDOW_SIZE));
        if (!window.add(key)) {
            return; // 窗口内重复，直接跳过
        }

        // 第二层：DB 唯一键兜底（冷路径）
        try {
            eventRepo.saveIfAbsent(key, record.timestamp(), record.value());
        } catch (DuplicateKeyException e) {
            return; // 窗口外重复（如 rebalance 后的历史消息）
        }

        // 业务处理
        OrderEvent event = deserialize(record.value());
        processOrder(event);
    }

    private OrderEvent deserialize(String json) {
        try { return mapper.readValue(json, OrderEvent.class); }
        catch (Exception e) { throw new IllegalArgumentException(e); }
    }

    private void processOrder(OrderEvent event) { /* 业务逻辑 */ }

    /** 线程安全 LRU 集合，用 LinkedHashMap 的 access-order 模式实现 */
    static class LruSet {
        private final Map<String, Boolean> map;
        private final int capacity;

        LruSet(int capacity) {
            this.capacity = capacity;
            this.map = java.util.Collections.synchronizedMap(
                new LinkedHashMap<String, Boolean>(capacity, 0.75f, true) {
                    @Override
                    protected boolean removeEldestEntry(Map.Entry<String, Boolean> e) {
                        return size() > capacity;
                    }
                });
        }

        boolean add(String key) {
            synchronized (map) {
                boolean isNew = !map.containsKey(key);
                map.put(key, Boolean.TRUE);
                return isNew;
            }
        }
    }
}

前置条件

Producer 端必须用 order_id 作为消息 key，保证同一订单的事件落到同一分区。Kafka 保证分区内顺序，因此单分区内 LRU 是有效的。

失效边界分析

| 场景 | 行为 | 风险等级 | |---|---|---| | Broker 宕机 | 重启后重新投递，内存窗口已清空 → 全量走 DB 兜底 | 安全，冷启动时 DB 压力短暂上升 | | rebalance | 分区重新分配，新消费者的 LRU 为空 → 窗口内消息走 DB | 安全，DB 兜底覆盖 | | 消费者 JVM 重启 | LRU 清空 → 同 Broker 宕机 | 安全 | | 同一分区内消息乱序 | Kafka 保证分区内有序，不会乱序 | 无风险（前提是 producer 用 order_id 做 key） | | LRU 窗口溢出 | 5000 TPS 下 10000 条窗口约覆盖 2s，超出窗口走 DB | 需评估窗口大小 vs DB 压力，建议压测确认 | | 网络分区 | 同方案一 | 安全 |

ShardingSphere 适配

• 第二层 DB 兜底同方案一，需分片键 = 唯一键首列。
• 核心优势：99%+ 的重复在内存拦截，DB 仅处理窗口外冷路径（rebalance / 重启后的历史消息），大幅降低跨分片查询频率，直接解决 P99 45ms 的痛点。
• 不引入新中间件，满足约束。

五、选型决策矩阵

| 维度 | 方案一：DB 唯一键 | 方案二：Redis SETNX | 方案三：分层去重 | |---|---|---|---| | 去重 P99 延迟 | 8~45ms（分片后更高） | 1~3ms | <0.1ms（热）/ 8~45ms（冷） | | 5000 TPS 吞吐 | 受限于 DB 连接池，约 2000~3000 TPS | Redis 轻松 10w+ TPS | 热路径无 DB 压力，DB 仅兜底 | | 一致性强度 | 强（DB 事务原子） | 中（依赖 AOF，failover 有 1~5s 窗口） | 强（DB 兜底 + 内存加速） | | 运维成本 | 低（无额外组件） | 中（需维护 Redis 高可用） | 中（需调 LRU 窗口大小） | | 故障恢复时间 | 快（DB 重启即恢复） | 中（Redis 重启后可重建） | 快（冷路径走 DB，内存重建无影响） | | ShardingSphere 兼容 | 需精确路由，跨分片慢 | 完全绕开 | 热路径绕开，冷路径同方案一 | | 不可丢消息保证 | 是 | 是（fail-safe） | 是 | | 实现复杂度 | 低 | 中 | 中高 |

六、推荐结论

推荐方案三（分层去重）作为主方案，方案二（Redis）作为可选增强层。

理由：

1. 你的核心痛点是"分片后去重查询慢"——方案三用内存 LRU 把 99%+ 的重复拦截在内存中，DB 只处理冷路径，直接解决 P99 45ms 的问题。
2. 方案三不引入新中间件，满足约束。
3. 如果未来 TPS 翻倍或需要更强去重保障，叠加 Redis SETNX 作为 LRU 和 DB 之间的"第二层缓存"，形成三级去重：内存 LRU → Redis → DB。
4. 方案一（纯 DB）作为兜底永远存在——无论用哪种加速层，DB 唯一键是最终一致性保障。

不推荐纯 Redis 方案作为唯一去重层：Redis 主从切换 + AOF 异步刷盘的组合在极端情况下有 1~5s 的数据丢失窗口，对于"不可丢消息"的要求不够安全。Redis 适合做加速层，不适合做唯一保障。

七、开放问题：Redis vs 纯 DB @ 5000 TPS 压测思路

压测环境

• Kafka：3 broker，8 分区，acks=all，min.insync.replicas=2
• MySQL：8 分库 × 16 分表，主从 + 半同步复制
• Redis：主从 + Sentinel，AOF everysec
• 消费者：4 实例，每实例 2 线程

压测工具与监控

# 生产者压测
kafka-producer-perf-test.sh \
  --broker-list broker1:9092,broker2:9092,broker3:9092 \
  --topic order-events \
  --num-records 5000000 \
  --record-size 512 \
  --producer-props acks=all \
  --throughput 5000

# 消费者 lag 监控（Prometheus）
# kafka_consumer_group_lag{group="order-svc"}

# DB 监控
mysql> SHOW GLOBAL STATUS LIKE 'Threads_connected';
mysql> SHOW GLOBAL STATUS LIKE 'Innodb_row_lock_waits';

# Redis 监控
redis-cli INFO stats | grep -E 'rejected_connections|total_commands'
redis-cli INFO persistence | grep -E 'aof_last_write|aof_delayed'

测试场景

1. 稳态：5000 TPS 持续 30min，记录 P50/P95/P99 延迟、DB QPS、Redis ops/s
2. 突发：10000 TPS × 5min，观察背压与消费 lag
3. Redis failover：redis-cli -h master DEL idem:test 后 kill 主节点，观察去重是否失效、恢复时间
4. DB 主从切换：mysql> CHANGE MASTER TO ...，观察事务回滚与重试
5. 消费者 kill -9 后重启：观察重复消费窗口时长（预期 200ms 内被 DB 唯一键兜底）

预期结果

| 指标 | Redis SETNX | 纯 DB 唯一键 | |---|---|---| | P99 去重延迟 | 1~3ms | 8~45ms | | 5000 TPS 下消费者 CPU | ~20% | ~45%（DB 等待占比高） | | Redis/DB 宕机时 | 消费停滞（fail-safe） | 不受影响（DB 就是自身） | | 30min 稳态错误率 | 0（AOF 保护） | 0 |

结论：5000 TPS 下 Redis 明显更稳（延迟低一个数量级），但必须配合 DB 唯一键兜底。纯 DB 在 5000 TPS 下不是"不可用"，而是"P99 贴近 50ms 红线"，没有余量应对突发。

八、最小依赖说明（Maven）

<dependencies>
    <!-- Kafka 消费者 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
    <!-- JPA + DB 唯一键 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <!-- Redis SETNX（方案二/增强层） -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <!-- ShardingSphere 分库分表 -->
    <dependency>
        <groupId>com.shardingsphere</groupId>
        <artifactId>shardingsphere-jdbc-core-spring-boot-starter</artifactId>
        <version>5.4.1</version>
    </dependency>
    <!-- MySQL 驱动 -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.33</version>
    </dependency>
</dependencies>

Kafka 3.x 客户端由 spring-kafka 2.8.x 管理，无需单独声明版本。Spring Boot 2.7.x 的 BOM 统一管理其余依赖版本。

一句话总结：在 ShardingSphere 分库分表场景下，"内存 LRU 热路径 + DB 唯一键冷路径"的分层去重是最优解——它把 99%+ 的重复拦截在 0.1ms 内，同时用 DB 唯一键保证最终一致性，不引入新中间件，且天然适配分片路由。