Spring Batch批量处理图片文件：分区方案疑问与实现求助

关于MultiResourcePartitioner创建400个分区的可行性 #

理论上这种方式能跑，但实际完全不推荐。每个文件对应一个分区会生成400个独立的Step执行实例，带来大量线程上下文切换、资源调度的开销，不仅不会提升效率，反而可能因为线程过载导致系统卡顿，甚至出现资源耗尽的情况。Spring Batch的分区更适合批量分组的场景，而非细粒度到单个文件。

按数量划分分区的实现方案 #

要实现10个分区、每个处理40张图片的需求，核心是自定义分区器和对应的读取器，以下是具体实现步骤：

1. 自定义按数量分组的Partitioner #

这个分区器会先扫描源目录收集所有符合格式的图片，再按指定大小拆分成分区，每个分区存储一组文件的路径列表。

import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class FileCountPartitioner implements Partitioner {

    private String sourceDirectory;
    private final List<String> allowedExtensions = List.of(".jpg", ".jpeg", ".heic");
    private int partitionSize = 40;

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        List<File> imageFiles = scanImageFiles();
        Map<String, ExecutionContext> partitions = new HashMap<>();

        int partitionIndex = 0;
        for (int i = 0; i < imageFiles.size(); i += partitionSize) {
            int end = Math.min(i + partitionSize, imageFiles.size());
            List<String> filePaths = imageFiles.subList(i, end).stream()
                    .map(File::getAbsolutePath)
                    .toList();

            ExecutionContext context = new ExecutionContext();
            context.put("filePaths", filePaths);
            partitions.put("partition-" + partitionIndex, context);
            partitionIndex++;
        }

        return partitions;
    }

    private List<File> scanImageFiles() {
        List<File> files = new ArrayList<>();
        File dir = new File(sourceDirectory);
        if (dir.isDirectory() && dir.listFiles() != null) {
            for (File file : dir.listFiles()) {
                String fileName = file.getName().toLowerCase();
                if (allowedExtensions.stream().anyMatch(fileName::endsWith)) {
                    files.add(file);
                }
            }
        }
        return files;
    }

    // 注入配置参数
    public void setSourceDirectory(String sourceDirectory) {
        this.sourceDirectory = sourceDirectory;
    }

    public void setPartitionSize(int partitionSize) {
        this.partitionSize = partitionSize;
    }
}

2. 配置分区作业的核心组件 #

包括读取器、处理器、写入器，以及主从Step的配置：

@Configuration
@EnableBatchProcessing
public class ImageBatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    // 自定义分区器实例
    @Bean
    public Partitioner fileCountPartitioner() {
        FileCountPartitioner partitioner = new FileCountPartitioner();
        partitioner.setSourceDirectory("/your/source/dir");
        partitioner.setPartitionSize(40);
        return partitioner;
    }

    // 并行任务执行器，线程数匹配分区数
    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(10);
        executor.setThreadNamePrefix("image-proc-");
        executor.initialize();
        return executor;
    }

    // 读取当前分区内的文件
    @Bean
    public ItemReader<File> imageFileReader() {
        return new ItemReader<File>() {
            private List<String> filePaths;
            private int currentIndex = 0;

            @Override
            public File read() {
                if (filePaths == null) {
                    ExecutionContext context = StepSynchronizationManager.getContext()
                            .getStepExecution().getExecutionContext();
                    filePaths = (List<String>) context.get("filePaths");
                }
                return currentIndex < filePaths.size() 
                        ? new File(filePaths.get(currentIndex++)) 
                        : null;
            }
        };
    }

    // 处理日期逻辑：对比文件名与EXIF日期，生成重命名请求
    @Bean
    public ItemProcessor<File, FileRenameRequest> imageDateProcessor() {
        return file -> {
            // 1. 从文件名提取日期（根据你的命名规则实现）
            String fileNameDate = extractDateFromFileName(file.getName());
            // 2. 读取EXIF元数据中的日期（可使用metadata-extractor库）
            String exifDate = extractExifDate(file);
            // 3. 取最早的日期作为新文件名的基础
            String earliestDate = getEarliestDate(fileNameDate, exifDate);
            // 4. 生成新文件名（注意避免重复，可加序号）
            String newFileName = generateNewFileName(earliestDate, file.getName());
            return new FileRenameRequest(file, new File("/your/target/dir/" + newFileName));
        };
    }

    // 执行文件移动/重命名
    @Bean
    public ItemWriter<FileRenameRequest> imageMoveWriter() {
        return items -> {
            for (FileRenameRequest request : items) {
                boolean success = request.getSource().renameTo(request.getTarget());
                if (!success) {
                    // 处理移动失败，比如日志记录或重试
                    throw new RuntimeException("Failed to move: " + request.getSource());
                }
            }
        };
    }

    // 从Step：处理单个分区内的文件
    @Bean
    public Step slaveStep() {
        return stepBuilderFactory.get("slaveStep")
                .<File, FileRenameRequest>chunk(1) // 单文件处理，也可调整批量大小
                .reader(imageFileReader())
                .processor(imageDateProcessor())
                .writer(imageMoveWriter())
                .build();
    }

    // 主Step：分区调度
    @Bean
    public Step masterStep() {
        return stepBuilderFactory.get("masterStep")
                .partitioner(slaveStep().getName(), fileCountPartitioner())
                .step(slaveStep())
                .taskExecutor(taskExecutor())
                .build();
    }

    // 作业定义
    @Bean
    public Job imageProcessingJob() {
        return jobBuilderFactory.get("imageProcessingJob")
                .start(masterStep())
                .build();
    }

    // 辅助类：封装文件重命名信息
    public static class FileRenameRequest {
        private final File source;
        private final File target;

        public FileRenameRequest(File source, File target) {
            this.source = source;
            this.target = target;
        }

        public File getSource() { return source; }
        public File getTarget() { return target; }
    }

    // 以下是需要你根据实际需求实现的工具方法
    private String extractDateFromFileName(String fileName) {
        // 示例：从"IMG_20231001_123456.jpg"提取"20231001"
        return "";
    }

    private String extractExifDate(File file) {
        // 使用metadata-extractor读取EXIF的DateTimeOriginal字段
        // 需引入依赖：com.drewnoakes:metadata-extractor:2.18.0
        return "";
    }

    private String getEarliestDate(String date1, String date2) {
        // 解析并对比两个日期，返回较早的那个
        return date1;
    }

    private String generateNewFileName(String date, String originalName) {
        // 生成唯一的新文件名，比如"20231001_001.jpg"
        return date + "_" + originalName;
    }
}

关键说明 #

• 分区数和线程数建议匹配CPU核心数（比如8核设8-10个分区），避免过度竞争资源。
• EXIF日期读取可使用metadata-extractor库，无需额外配置，直接解析图片文件即可。
• 文件名去重逻辑需根据实际情况实现，比如在新文件名后加递增序号，避免目标目录文件冲突。

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