Spring Batch中如何按优先级处理同用户的图片上传任务？

你的需求核心是对同一Person的图片按优先级依次尝试上传，成功后终止该组后续项的处理，默认逐行处理的Processor无法直接实现，这里提供两种可行方案：

方法一：自定义分组Reader，将同Person数据作为一个整体传入Processor #

先在Reader阶段把同一Person的所有图片数据归为一组，按优先级排序后作为单个Item返回，再在Processor中遍历组内项尝试上传，成功即停止。

1. 自定义分组Reader示例 #

public class GroupedPersonImageReader implements ItemReader<List<PersonImage>> {
    private final ItemReader<PersonImage> delegateReader;
    private List<PersonImage> currentGroup;
    private String currentPersonId;

    @Override
    public List<PersonImage> read() throws Exception {
        if (currentGroup == null) {
            currentGroup = new ArrayList<>();
            PersonImage item = delegateReader.read();
            if (item == null) {
                return null;
            }
            currentPersonId = item.getPersonId();
            currentGroup.add(item);
            
            // 读取同Person的所有剩余项
            while ((item = delegateReader.read()) != null && currentPersonId.equals(item.getPersonId())) {
                currentGroup.add(item);
            }
            
            // 按优先级升序排序（Priority1在前）
            currentGroup.sort(Comparator.comparingInt(PersonImage::getPriority));
        }
        List<PersonImage> groupToProcess = currentGroup;
        currentGroup = null;
        return groupToProcess;
    }
}

2. 对应Processor实现 #

public class PersonImageUploadProcessor implements ItemProcessor<List<PersonImage>, Boolean> {
    private final AmazonS3 s3Client;
    private static final Logger log = LoggerFactory.getLogger(PersonImageUploadProcessor.class);

    @Override
    public Boolean process(List<PersonImage> imageGroup) throws Exception {
        for (PersonImage image : imageGroup) {
            try {
                // 执行S3上传逻辑
                s3Client.putObject("your-bucket-name", image.getImageS3Key(), new File(image.getLocalImagePath()));
                log.info("Person {}的图片{}上传成功，终止后续尝试", image.getPersonId(), image.getImageName());
                return true;
            } catch (Exception e) {
                log.error("Person {}的图片{}上传失败，尝试下一张", image.getPersonId(), image.getImageName(), e);
                continue;
            }
        }
        log.warn("Person {}的所有图片均上传失败", imageGroup.get(0).getPersonId());
        return false;
    }
}

方法二：借助ExecutionContext记录上传状态，跳过已成功组的后续项 #

如果不想修改Reader结构，可以在Processor中通过StepExecution的ExecutionContext记录同一Person的上传状态，已成功则直接跳过当前项。

实现示例 #

public class PriorityImageUploadProcessor implements ItemProcessor<PersonImage, PersonImage>, StepExecutionListener {
    private final AmazonS3 s3Client;
    private StepExecution stepExecution;
    private static final String UPLOAD_SUCCESS_MARK = "UPLOAD_SUCCESS_";
    private static final Logger log = LoggerFactory.getLogger(PriorityImageUploadProcessor.class);

    @Override
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public PersonImage process(PersonImage item) throws Exception {
        String successKey = UPLOAD_SUCCESS_MARK + item.getPersonId();
        // 检查当前Person是否已有成功上传记录
        if (stepExecution.getExecutionContext().containsKey(successKey)) {
            return null; // 返回null让Writer跳过当前项
        }

        try {
            s3Client.putObject("your-bucket-name", item.getImageS3Key(), new File(item.getLocalImagePath()));
            stepExecution.getExecutionContext().put(successKey, true);
            log.info("Person {}的图片{}上传成功", item.getPersonId(), item.getImageName());
            return item;
        } catch (Exception e) {
            log.error("Person {}的图片{}上传失败", item.getPersonId(), item.getImageName(), e);
            return null; // 上传失败，跳过当前项，继续下一个
        }
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return null;
    }
}

关键注意事项 #

• 两种方案都要求Reader返回的数据先按Person分组，再按优先级从高到低排序，比如用SQL查询时加上ORDER BY person_id, priority ASC。
• 方法一中的分组Reader要注意边界处理，避免遗漏最后一组数据。
• 方法二的ExecutionContext会在Step执行期间保留状态，适合单Step内的同组数据处理。

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