Spring Integration：如何将目录所有文件读取为单个Message<List<File>>

方式一：基于Spring Integration原生组件批量读取 #

如果使用Spring Integration，可以通过包装原生的FileReadingMessageSource，实现一次性收集目录下所有文件并封装为Message<List<File>>：

@Bean
public MessageSource<List<File>> batchFileSource() {
    FileReadingMessageSource singleFileSource = new FileReadingMessageSource();
    singleFileSource.setDirectory(new File("/your/target/directory"));
    // 可替换为自定义过滤器，比如按文件后缀筛选
    singleFileSource.setFilter(new AcceptAllFileListFilter<>());

    return () -> {
        List<File> allFiles = new ArrayList<>();
        File file;
        // 循环读取直到目录下无未读取文件
        while ((file = singleFileSource.receive()) != null) {
            allFiles.add(file);
        }
        return MessageBuilder.withPayload(allFiles).build();
    };
}

这个Bean会在每次轮询时收集目标目录下的所有文件，生成包含文件列表的消息，后续可直接用于提取文件名生成状态报告。

方式二：自定义MessageSource实现（灵活扩展） #

如果需要递归遍历子目录、自定义文件筛选规则等个性化逻辑，可自行实现MessageSource<List<File>>接口：

1. 实现自定义MessageSource类 #

public class BatchDirectoryFileSource implements MessageSource<List<File>> {

    private final File targetDir;
    private final boolean includeSubDirs;
    // 可选：记录已读取文件，避免重复处理
    private final Set<String> processedFiles = new HashSet<>();

    public BatchDirectoryFileSource(String dirPath, boolean includeSubDirs) {
        this.targetDir = new File(dirPath);
        this.includeSubDirs = includeSubDirs;
        if (!targetDir.exists() || !targetDir.isDirectory()) {
            throw new IllegalArgumentException("无效的目录路径: " + dirPath);
        }
    }

    @Override
    public Message<List<File>> receive() {
        List<File> files = collectFiles(targetDir);
        // 过滤已处理文件（可选逻辑）
        List<File> newFiles = files.stream()
                .filter(file -> processedFiles.add(file.getAbsolutePath()))
                .collect(Collectors.toList());
        return MessageBuilder.withPayload(newFiles).build();
    }

    private List<File> collectFiles(File dir) {
        List<File> fileList = new ArrayList<>();
        File[] files = dir.listFiles();
        if (files == null) return fileList;

        for (File file : files) {
            if (file.isFile()) {
                fileList.add(file);
            } else if (includeSubDirs && file.isDirectory()) {
                fileList.addAll(collectFiles(file));
            }
        }
        return fileList;
    }
}

2. 注册为Spring Bean #

@Configuration
public class FileSourceConfig {

    @Bean
    public MessageSource<List<File>> customBatchFileSource() {
        // 参数：目标目录路径，是否包含子目录
        return new BatchDirectoryFileSource("/your/target/directory", true);
    }
}

3. 生成并发送状态报告 #

将自定义的MessageSource接入消息流，提取文件名生成报告：

@Bean
public IntegrationFlow fileStatusReportFlow() {
    return IntegrationFlows.from(customBatchFileSource(), 
            spec -> spec.poller(Pollers.fixedDelay(10000))) // 10秒轮询一次
            .transform((List<File> files) -> {
                StringBuilder report = new StringBuilder("目录文件状态报告:\n");
                files.forEach(file -> report.append("- 文件名: ").append(file.getName()).append("\n"));
                return report.toString();
            })
            // 替换为实际的报告发送逻辑（如邮件、MQ推送等）
            .handle(report -> System.out.println("发送状态报告:\n" + report.getPayload()))
            .get();
}

注意事项 #

• 若需避免重复处理文件，可在自定义Source中添加已处理文件的跟踪逻辑（如示例中的processedFiles集合）
• 可根据业务需求调整轮询间隔、文件过滤规则（如添加后缀名筛选）
• 生产环境建议添加目录存在性、权限校验等异常处理逻辑

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