Java并行流处理实体列表时重复执行问题排查与解决

可能的原因分析 #

结合你的代码和依赖配置，几个常见的触发点：

• Spring Retry自动重试：你的依赖里引入了spring-retry，如果forEach中调用的方法标记了@Retryable，一旦执行出错就会自动重试，导致同一个实体被多次处理。
• 惰性加载集合重复遍历：如果entityList是JPA延迟查询的结果（即使声明为List），每次遍历Stream都会重新触发数据库查询，拿到重复的实体数据。
• 异步任务重复提交：你用customThreadPool.submit()提交异步任务，如果外部逻辑重复触发这段代码，会导致同一批实体被多次处理。

具体修复方案 #

1. 控制重试逻辑 #

如果是Spring Retry导致的重复执行：

• 直接移除业务方法上的@Retryable注解（如果不需要重试）；
• 若必须保留重试，给重试逻辑添加幂等性校验，比如在处理前检查实体状态是否已变更，避免重复执行。

2. 固化实体列表为内存集合 #

将查询结果转换成普通内存集合，避免惰性加载带来的重复查询：

// 用ArrayList包裹，确保一次性加载所有数据到内存
List<Entity> entityList = new ArrayList<>(repository.findTop1000ByStatus(PENDING));

3. 正确使用自定义线程池 #

改用invoke()替代submit()，确保任务执行完成后再继续后续逻辑，同时用try-with-resources自动关闭线程池：

List<Entity> entityList = new ArrayList<>(repository.findTop1000ByStatus(PENDING));
int poolSize = Math.min(entityList.size(), 100);
// 自动关闭线程池
try (ForkJoinPool customThreadPool = new ForkJoinPool(poolSize)) {
    // invoke()会阻塞直到所有任务执行完毕
    customThreadPool.invoke(() -> entityList.parallelStream().forEach(entity -> {
        long userId = entity.getUserId();
        UserDto user;
        try {
            // 你的业务逻辑
        } catch (Exception e) {
            // 异常处理，避免重试连锁触发
            log.error("处理实体ID:{}失败", entity.getId(), e);
        }
    }));
}

4. 强制去重处理 #

如果上述方案仍无法解决，可在Stream逻辑中添加已处理ID的校验，确保每个实体仅执行一次：

// 线程安全的集合记录已处理ID
Set<Long> processedIds = ConcurrentHashMap.newKeySet();
entityList.parallelStream().forEach(entity -> {
    // add()返回false表示ID已存在，跳过处理
    if (processedIds.add(entity.getId())) {
        // 你的业务逻辑
    }
});

额外排查建议 #

在forEach逻辑开头添加日志，打印entity.getId()，快速定位被重复处理的实体，进一步缩小问题范围。

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