本文目录导读：
在Java分布式系统中实现数据清理API的过期机制,通常有几种主流方案，下面我会从简单到复杂，结合实际场景给出具体实现思路。
基于TTL（Time To Live）的过期策略
Redis TTL实现（最常用）
@Service
public class CacheCleanupService {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/**
* 设置带过期时间的数据
* @param key 缓存键
* @param value 数据
* @param timeout 过期时间
* @param unit 时间单位
*/
public void setWithExpiration(String key, Object value, long timeout, TimeUnit unit) {
redisTemplate.opsForValue().set(key, value, timeout, unit);
}
/**
* 批量设置过期时间
*/
public void batchExpire(List<String> keys, long timeout, TimeUnit unit) {
keys.forEach(key -> redisTemplate.expire(key, timeout, unit));
}
}
自定义注解实现
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface DataCleanup {
long ttl() default 3600; // 默认1小时
TimeUnit unit() default TimeUnit.SECONDS;
}
@Aspect
@Component
public class DataCleanupAspect {
@Around("@annotation(cleanup)")
public Object handleCleanup(ProceedingJoinPoint joinPoint, DataCleanup cleanup) {
// 执行前检查
String key = generateKey(joinPoint);
if (isExpired(key)) {
// 清理过期数据
cleanupExpiredData(key);
}
Object result = joinPoint.proceed();
// 执行后设置过期时间
setExpiration(key, cleanup.ttl(), cleanup.unit());
return result;
}
}
分布式任务调度清理
使用Quartz + ZooKeeper实现分布式调度
@Component
public class DistributedCleanupScheduler {
@Autowired
private CuratorFramework curatorClient;
@Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
public void scheduledCleanup() {
// 分布式锁，确保只有一个节点执行
InterProcessMutex lock = new InterProcessMutex(curatorClient, "/cleanup/lock");
try {
if (lock.acquire(10, TimeUnit.SECONDS)) {
// 执行清理逻辑
executeCleanup();
}
} catch (Exception e) {
log.error("分布式清理任务执行失败", e);
} finally {
try {
lock.release();
} catch (Exception e) {
log.error("释放分布式锁失败", e);
}
}
}
private void executeCleanup() {
// 1. 查询需要清理的数据
List<CleanupTask> tasks = cleanupTaskRepository.findExpiredTasks();
// 2. 批量处理
tasks.parallelStream().forEach(task -> {
try {
cleanupService.cleanup(task.getDataKey());
task.setStatus(CleanupStatus.COMPLETED);
} catch (Exception e) {
task.setStatus(CleanupStatus.FAILED);
log.error("清理任务执行失败: {}", task.getId(), e);
}
});
// 3. 更新任务状态
cleanupTaskRepository.saveAll(tasks);
}
}
基于时间戳的惰性删除
数据实体设计
@Entity
@Table(name = "data_entity")
public class DataEntity {
@Id
private String id;
private String dataKey;
@Column(name = "expire_time")
private LocalDateTime expireTime;
@Column(name = "is_deleted")
private boolean isDeleted;
// 查询时自动过滤过期数据
@PrePersist
public void prePersist() {
if (expireTime == null) {
expireTime = LocalDateTime.now().plusDays(30); // 默认30天过期
}
isDeleted = false;
}
// 检查是否过期
public boolean isExpired() {
return LocalDateTime.now().isAfter(expireTime);
}
}
清理服务实现
@Service
public class LazyCleanupService {
@Autowired
private DataEntityRepository repository;
/**
* 查询有效数据（自动过滤过期）
*/
public List<DataEntity> findValidData(String condition) {
return repository.findByConditionAndExpireTimeAfter(condition, LocalDateTime.now());
}
/**
* 访问时检查并清理过期数据
*/
public DataEntity accessData(String dataKey) {
DataEntity entity = repository.findByDataKey(dataKey);
if (entity != null && entity.isExpired()) {
// 惰性删除：访问时发现过期，立即清理
repository.delete(entity);
return null;
}
return entity;
}
}
基于事件驱动的清理
使用消息队列实现异步清理
@Component
public class EventDrivenCleanup {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 发布清理事件
*/
public void publishCleanupEvent(String dataKey, long delay) {
CleanupEvent event = new CleanupEvent();
event.setDataKey(dataKey);
event.setTimestamp(System.currentTimeMillis());
// 发送延迟消息
rabbitTemplate.convertAndSend(
"cleanup.exchange",
"cleanup.route",
event,
message -> {
message.getMessageProperties().setDelay((int) delay);
return message;
}
);
}
/**
* 处理清理事件
*/
@RabbitListener(queues = "cleanup.queue")
public void handleCleanupEvent(CleanupEvent event) {
// 执行清理逻辑
cleanupService.cleanup(event.getDataKey());
// 记录清理日志
log.info("数据清理完成: key={}, timestamp={}",
event.getDataKey(), event.getTimestamp());
}
}
完整的过期策略配置
配置类设计
@Configuration
@ConfigurationProperties(prefix = "data.cleanup")
public class CleanupConfig {
// 默认过期时间
private long defaultTtl = 86400; // 24小时
// 不同数据类型的过期策略
private Map<String, CleanupPolicy> policies = new HashMap<>();
// 批量清理大小
private int batchSize = 1000;
// 清理线程池配置
private ExecutorConfig executor = new ExecutorConfig();
@Data
public static class CleanupPolicy {
private long ttl;
private TimeUnit unit = TimeUnit.SECONDS;
private boolean lazyCleanup = true;
private boolean scheduleCleanup = false;
private String cronExpression;
}
@Data
public static class ExecutorConfig {
private int corePoolSize = 10;
private int maxPoolSize = 50;
private int queueCapacity = 1000;
}
}
最佳实践建议
选择合适的过期策略
public class CleanupStrategyFactory {
private final Map<String, CleanupStrategy> strategies;
public CleanupStrategy getStrategy(DataCleanupRequest request) {
switch (request.getStrategyType()) {
case "TTL":
return new TtlCleanupStrategy(redisTemplate);
case "LAZY":
return new LazyCleanupStrategy(repository);
case "SCHEDULE":
return new ScheduleCleanupStrategy(scheduler);
default:
throw new IllegalArgumentException("不支持的清理策略: " + request.getStrategyType());
}
}
}
监控和日志
@Slf4j
@Component
public class CleanupMonitor {
private final MeterRegistry meterRegistry;
public void recordCleanup(CleanupResult result) {
// 记录清理数量
meterRegistry.counter("cleanup.count",
"type", result.getType(),
"status", result.getStatus()
).increment(result.getCount());
// 记录清理耗时
meterRegistry.timer("cleanup.duration",
"type", result.getType()
).record(result.getDuration(), TimeUnit.MILLISECONDS);
log.info("清理完成: type={}, count={}, duration={}ms",
result.getType(), result.getCount(), result.getDuration());
}
}
选择合适的过期策略需要考虑：
- 数据量大小：大数据量建议用批量调度
- 实时性要求：高实时性用TTL，允许延迟用惰性删除
- 一致性要求：强一致性用分布式锁，最终一致性用事件驱动
- 资源消耗：定期调度消耗固定资源，惰性删除按需消耗
建议组合使用多种策略,TTL作为主要过期机制，定期调度作为兜底清理，惰性删除作为最终保障。
