Spring Batch Integration多文件SFTP并行上传及重试配置需求

我来帮你梳理下如何用Spring Batch的原生能力替换掉原来的Future+ThreadPool方案，实现多文件SFTP并行上传，同时加上失败重试的逻辑：

核心思路 #

• 用Spring Batch的**分区Step（Partitioned Step）**实现Tasklet的并行执行：通过分区器拆分待上传文件列表，每个分区对应一个独立的上传Tasklet
• 结合Spring Batch的重试注解，为上传Tasklet配置自定义重试策略，实现失败后按指定间隔重试

1. 配置文件分区器（Partitioner） #

分区器负责将待上传文件拆分为多个分片，每个分片作为独立任务并行执行：

@Bean
public Partitioner filePartitioner() {
    return new Partitioner() {
        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            // 实际业务中从目录/数据库获取待上传文件路径列表
            List<String> filePaths = getPendingUploadFiles();
            Map<String, ExecutionContext> partitions = new HashMap<>();
            
            int index = 0;
            for (String filePath : filePaths) {
                ExecutionContext context = new ExecutionContext();
                context.putString("targetFilePath", filePath);
                partitions.put("upload-partition-" + index++, context);
            }
            return partitions;
        }
    };
}

// 模拟获取待上传文件列表的方法
private List<String> getPendingUploadFiles() {
    return Arrays.asList("/local/files/file1.txt", "/local/files/file2.jpg", "/local/files/file3.pdf");
}

2. 实现带重试逻辑的上传Tasklet #

这个Tasklet调用你的UploadGateway执行单个文件上传，通过注解配置重试规则：

@Component
@EnableRetry // 需在启动类或配置类添加此注解开启重试功能
public class SftpUploadTasklet implements Tasklet {

    private static final Logger log = LoggerFactory.getLogger(SftpUploadTasklet.class);
    
    @Autowired
    private UploadGateway gateway;

    @Override
    @Retryable(
            value = {SftpException.class, IOException.class}, // 指定触发重试的异常类型
            maxAttempts = 3, // 最大重试次数
            backoff = @Backoff(delay = 2000, multiplier = 1.5) // 初始延迟2秒，后续延迟按1.5倍递增
    )
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        String filePath = chunkContext.getStepContext().getJobExecutionContext().getString("targetFilePath");
        File targetFile = new File(filePath);
        
        // 调用Spring Integration网关执行SFTP上传
        boolean uploadResult = gateway.uploadFile(targetFile);
        
        if (!uploadResult) {
            throw new SftpException("文件上传失败：" + targetFile.getName());
        }
        log.info("文件{}上传成功", targetFile.getName());
        return RepeatStatus.FINISHED;
    }

    // 重试耗尽后的兜底处理方法
    @Recover
    public RepeatStatus handleUploadFailure(Exception e, StepContribution contribution, ChunkContext chunkContext) {
        String filePath = chunkContext.getStepContext().getJobExecutionContext().getString("targetFilePath");
        log.error("文件{}重试3次后仍上传失败，终止操作", filePath, e);
        // 这里可扩展：记录失败日志、发送告警、标记文件状态等
        return RepeatStatus.FINISHED;
    }
}

3. 配置并行执行的Job和Step #

通过分区Step串联所有组件，实现多文件并行上传：

@Autowired
private JobBuilderFactory jobBuilderFactory;

@Autowired
private StepBuilderFactory stepBuilderFactory;

@Autowired
private SftpUploadTasklet sftpUploadTasklet;

@Autowired
private Partitioner filePartitioner;

@Bean
public Job multiFileSftpUploadJob() {
    return jobBuilderFactory.get("multiFileSftpUploadJob")
            .start(partitionedUploadStep())
            .build();
}

@Bean
public Step partitionedUploadStep() {
    return stepBuilderFactory.get("partitionedUploadStep")
            .partitioner("singleFileUploadStep", filePartitioner)
            .step(singleFileUploadStep())
            .gridSize(5) // 并行执行的任务数，根据服务器资源调整
            .taskExecutor(uploadTaskExecutor()) // 指定自定义线程池
            .build();
}

@Bean
public Step singleFileUploadStep() {
    return stepBuilderFactory.get("singleFileUploadStep")
            .tasklet(sftpUploadTasklet)
            .build();
}

@Bean
public TaskExecutor uploadTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(20);
    executor.setThreadNamePrefix("sftp-upload-thread-");
    executor.initialize();
    return executor;
}

4. 整合Spring Integration的SFTP上传网关 #

确保你的UploadGateway已正确配置SFTP出站通道：

@MessagingGateway
public interface UploadGateway {
    @Gateway(requestChannel = "sftpUploadChannel")
    boolean uploadFile(File file);
}

// SFTP通道与处理器配置
@Bean
public MessageChannel sftpUploadChannel() {
    return new DirectChannel();
}

@Bean
@ServiceActivator(inputChannel = "sftpUploadChannel")
public MessageHandler sftpMessageHandler(SftpSessionFactory sftpSessionFactory) {
    SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory);
    handler.setRemoteDirectoryExpression(new LiteralExpression("/sftp/server/upload/path"));
    handler.setFileNameGenerator(message -> ((File) message.getPayload()).getName());
    return handler;
}

@Bean
public SftpSessionFactory sftpSessionFactory() {
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
    factory.setHost("your-sftp-host");
    factory.setPort(22);
    factory.setUser("sftp-username");
    factory.setPassword("sftp-password");
    // 可选：密钥认证方式
    // factory.setPrivateKey(new ClassPathResource("private-key.pem"));
    // factory.setPrivateKeyPassphrase("key-passphrase");
    factory.setAllowUnknownKeys(true);
    return factory;
}

关键优势说明 #

• 规范的并行执行：用Spring Batch分区Step替代手动线程池管理，更符合Spring生态规范，自带任务监控、失败追踪能力
• 灵活的重试策略：通过注解轻松配置重试次数、延迟规则，支持指数退避，无需手动编写重试逻辑
• 清晰的职责划分：分区器、Tasklet、Job各司其职，代码更易维护和扩展

内容的提问来源于stack exchange，提问作者A.S.karthick