本文介绍一种不依赖 Quartz 的纯 Spring 原生方案，通过分布式锁 + 数据库乐观更新机制，在多节点集群中安全、可靠地执行定时轮询任务，确保每条待处理记录仅被一个实例处理一次。
本文介绍一种不依赖 Quartz 的纯 Spring 原生方案，通过分布式锁 + 数据库乐观更新机制，在多节点集群中安全、可靠地执行定时轮询任务，确保每条待处理记录仅被一个实例处理一次。
在 Spring 生态中，若需在集群环境下实现“每 5 分钟扫描一次数据库中 scheduled_time ≤ now 的待处理记录，并确保每条记录仅被一个服务实例消费一次”，直接使用 @Scheduled 是不够的——因为该注解默认是单机生效，多个实例会并发执行相同逻辑，导致重复处理。因此，必须引入协调机制。
✅ 核心设计思路
-
定时触发层：仍用
@Scheduled(fixedDelay = 300_000)每 5 分钟唤醒一次； - 竞争控制层：采用「数据库行级更新 + 影子状态字段」实现轻量分布式锁（无额外中间件依赖）；
-
幂等保障层：为每条记录增加
status字段（如'PENDING'/'PROCESSING'/'PROCESSED'），仅更新status = 'PENDING' AND scheduled_time 的记录为 <code>'PROCESSING'，并检查UPDATE返回影响行数 > 0 才真正执行业务逻辑； -
失败兜底：对长时间处于
'PROCESSING'状态的记录设置超时重试机制（如last_updated_at ）。
? 示例实现（Spring Boot + JPA）
@Service
public class ScheduledRecordProcessor {
@Scheduled(fixedDelay = 300_000) // 每5分钟执行一次
public void processDueRecords() {
LocalDateTime now = LocalDateTime.now();
// Step 1: 尝试抢占最多10条待处理记录（避免单次负载过大）
int updated = recordRepository.markAsProcessing(now, 10);
if (updated == 0) return;
// Step 2: 查询刚标记为 PROCESSING 的记录（确保一致性读）
List<ScheduledRecord> records = recordRepository.findByStatusAndProcessedTime(
RecordStatus.PROCESSING, now.minusMinutes(1)); // 宽松时间窗口防时钟漂移
for (ScheduledRecord r : records) {
try {
executeBusinessLogic(r);
r.setStatus(RecordStatus.PROCESSED);
recordRepository.save(r);
} catch (Exception e) {
log.error("Failed to process record {}", r.getId(), e);
r.setStatus(RecordStatus.FAILED);
recordRepository.save(r);
}
}
}
private void executeBusinessLogic(ScheduledRecord record) {
// 实际业务逻辑：发消息、调接口、更新关联数据等
}
}对应 Repository 方法（JPA + 自定义 SQL）：
@Repository
public interface RecordRepository extends JpaRepository<ScheduledRecord, Long> {
@Modifying
@Query("UPDATE ScheduledRecord r " +
"SET r.status = :processing, r.lastUpdatedAt = CURRENT_TIMESTAMP " +
"WHERE r.status = :pending " +
" AND r.scheduledTime <= :now " +
" AND r.id IN (" +
" SELECT r2.id FROM ScheduledRecord r2 " +
" WHERE r2.status = :pending AND r2.scheduledTime <= :now " +
" ORDER BY r2.scheduledTime ASC " +
" LIMIT :limit" +
")")
int markAsProcessing(@Param("now") LocalDateTime now,
@Param("pending") RecordStatus pending,
@Param("processing") RecordStatus processing,
@Param("limit") int limit);
List<ScheduledRecord> findByStatusAndProcessedTime(
RecordStatus status, LocalDateTime cutoffTime);
}⚠️ 注意事项：
- 必须为
status和scheduled_time字段建立联合索引（如INDEX idx_status_time (status, scheduled_time)），否则轮询性能将随数据增长急剧下降；- 若数据库不支持
LIMIT在UPDATE中（如 MySQL 5.7+ 支持，PostgreSQL 需用 CTE），可改用SELECT ... FOR UPDATE SKIP LOCKED（推荐）；- 不建议在
@Scheduled方法中做耗时操作或阻塞 I/O，应异步化或拆分批次；- 启动类需添加
@EnableScheduling，且确保 Spring 容器已加载该@ServiceBean。
该方案零外部依赖、部署简单、可观测性强，适用于中小规模集群（≤10 节点）及对强一致性要求不极致但需高可用的定时调度场景。
