如何使用TransactionSynchronizationFactory构造SpEL实现FTP远程文件跨目录迁移

要实现事务提交后将FTP远程文件从/in移动到/archive目录，核心是利用FtpRemoteFileTemplate封装的FTP操作，结合SpEL表达式在事务提交后执行移动逻辑。以下是具体实现步骤：

1. 配置FtpRemoteFileTemplate Bean #

这个Bean封装了FTP服务器的文件操作方法，用于在SpEL中直接调用：

@Bean
public FtpRemoteFileTemplate ftpRemoteFileTemplate(SessionFactory<FTPFile> ftpSessionFactory) {
    return new FtpRemoteFileTemplate(ftpSessionFactory);
}

2. 配置事务同步处理器并构造SpEL表达式 #

通过ExpressionEvaluatingTransactionSynchronizationProcessor的setAfterCommitExpression方法，编写SpEL调用FtpRemoteFileTemplate的move方法。Spring Integration会将远程文件的元数据存入Message Header，可直接通过headers获取：

@Bean
public ExpressionEvaluatingTransactionSynchronizationProcessor ftpSyncProcessor(FtpRemoteFileTemplate ftpTemplate) {
    ExpressionEvaluatingTransactionSynchronizationProcessor processor = new ExpressionEvaluatingTransactionSynchronizationProcessor();
    
    // SpEL表达式：拼接源路径和目标路径，执行远程文件移动
    String moveExpr = """
        @ftpRemoteFileTemplate.move(
            headers['file_remoteDirectory'] + headers['file_remoteFile'].getName(),
            '/archive/' + headers['file_remoteFile'].getName()
        )
    """;
    processor.setAfterCommitExpression(new SpelExpressionParser().parseExpression(moveExpr));
    
    return processor;
}

表达式细节： #

• @ftpRemoteFileTemplate：通过SpEL的Bean引用语法，直接调用容器中的FtpRemoteFileTemplate实例
• headers['file_remoteDirectory']：获取远程文件所在的源目录（即配置的/in）
• headers['file_remoteFile']：获取远程文件的FTPFile对象，调用getName()获取文件名

3. 关联同步处理器到FTP入站适配器 #

在@InboundChannelAdapter中配置事务轮询器，并绑定事务同步处理器，确保只有事务提交后才执行移动操作：

@Bean
@InboundChannelAdapter(
    channel = "ftpInputChannel",
    poller = @Poller(fixedDelay = "5000", transactional = @Transactional)
)
public MessageSource<File> ftpInboundAdapter(
    SessionFactory<FTPFile> ftpSessionFactory,
    ExpressionEvaluatingTransactionSynchronizationProcessor ftpSyncProcessor
) {
    FtpInboundChannelAdapter adapter = new FtpInboundChannelAdapter(ftpSessionFactory);
    adapter.setRemoteDirectory("/in"); // 远程源目录
    adapter.setLocalDirectory(new File("./temp-ftp")); // 本地临时下载目录
    adapter.setAutoCreateLocalDirectory(true);
    adapter.setTransactionSynchronizationProcessor(ftpSyncProcessor); // 绑定同步处理器
    
    // 可选：配置文件过滤，避免重复处理
    adapter.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftp-inbound"));
    
    return adapter;
}

注意事项 #

• 确保FTP服务器上的/archive目录存在，若需自动创建，可在SpEL中添加目录创建逻辑：

  String moveExpr = """
      @ftpRemoteFileTemplate.mkdir('/archive');
      @ftpRemoteFileTemplate.move(
          headers['file_remoteDirectory'] + headers['file_remoteFile'].getName(),
          '/archive/' + headers['file_remoteFile'].getName()
      )
  """;
  

• 事务配置必须生效：@Poller的transactional属性需设置为@Transactional，否则afterCommit表达式不会触发
• 若需处理目标目录重名文件，可在目标文件名前添加时间戳等唯一标识：

  '/archive/' + T(java.time.LocalDateTime).now().format(T(java.time.format.DateTimeFormatter).ofPattern('yyyyMMddHHmmss')) + '_' + headers['file_remoteFile'].getName()
  

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