适用于基于ID和时间戳去重的分布式存储方案选型咨询

一、DynamoDB方案落地实现 #

你提出的DynamoDB方案完全适配当前需求，可按以下方式设计和实现：

1. 表结构定义 #

• 分区键：key（字符串类型，确保同一key的记录落在同一个分区）
• 排序键：timestamp（数字类型，存储毫秒级时间戳，支持范围查询）
• 启用TTL功能：为表添加expire_at属性，将其设为TTL字段，值为timestamp + 3600000（即记录生成1小时后自动删除），避免存储无限膨胀

2. 核心校验逻辑 #

针对传入的key和targetTimestamp，执行范围查询判断是否存在符合条件的记录：

// 初始化DynamoDB客户端（建议使用单例模式）
AmazonDynamoDB client = AmazonDynamoDBClientBuilder.defaultClient();
Table table = Table.loadTable(client, "duplicate_check_table");

// 构造查询条件，检查前后1小时内的记录
QuerySpec querySpec = new QuerySpec()
    .withKeyConditionExpression("key = :key_val AND timestamp BETWEEN :start_ts AND :end_ts")
    .withValueMap(new ValueMap()
        .withString(":key_val", key)
        .withNumber(":start_ts", timestamp - 3600000L)
        .withNumber(":end_ts", timestamp + 3600000L))
    .withLimit(1); // 找到一条记录即停止查询，减少传输开销

// 执行查询并判断结果
ItemCollection<QueryOutcome> result = table.query(querySpec);
boolean isDuplicate = result.iterator().hasNext();

// 若不存在重复，将当前记录写入表
if (!isDuplicate) {
    long expireAt = timestamp + 3600000L;
    table.putItem(new Item()
        .withString("key", key)
        .withNumber("timestamp", timestamp)
        .withNumber("expire_at", expireAt));
}

return isDuplicate;

3. 性能优化措施 #

• 二级本地缓存：在Kafka Streams应用侧引入Caffeine或Guava Cache，缓存最近1小时内的key-timestamp记录，缓存过期时间设为1小时。优先查询本地缓存，命中则直接返回结果，未命中再查询DynamoDB，可将响应时间降低至1ms以内
• 按需容量模式：开启DynamoDB按需付费模式，自动适配业务流量波动，避免预配置容量的浪费或不足
• 连接池优化：配置合理的DynamoDB客户端连接池大小，减少连接建立开销

二、备选方案对比 #

如果DynamoDB不符合你的技术栈偏好，可考虑以下方案：

• Redis Sorted Set：以key为集合名，timestamp为分数。使用ZRANGEBYSCORE key [targetTs-3600000] [targetTs+3600000] LIMIT 0 1判断是否存在重复，写入时执行ZADD key timestamp timestamp。Redis的低延迟特性能满足响应时间要求，但需部署集群保证高可用性，并设置EXPIRE自动清理过期集合
• Cassandra：以key为分区键，timestamp为聚类列，执行SELECT * FROM duplicate_check WHERE key = ? AND timestamp >= ? AND timestamp <= ? LIMIT 1。Cassandra适合海量数据存储，读写性能稳定，但需要配置TTL自动清理过期数据

三、关键注意事项 #

• 时间戳一致性：确保Kafka Streams应用和外部存储统一使用毫秒级时间戳，避免时区转换或精度差异导致的判断错误
• 并发写入处理：同一key同一时间戳的并发写入无需额外控制，因为只要存在记录就判定为重复，重复插入不影响校验结果；若需避免冗余存储，可使用DynamoDB的条件写入（ConditionExpression: attribute_not_exists(timestamp)）
• 监控与调优：监控外部存储的查询延迟、读写吞吐量，根据实际流量调整资源配置，确保响应时间符合要求

内容的提问来源于stack exchange，提问作者Kohei Nozaki