Spring Integration SFTP处理失败时如何从RedisMetadataStore移除文件？

核心结论 #

直接调用RedisMetadataStore的remove方法是合理的，只要确保使用和SftpPersistentAcceptOnceFileListFilter生成的一致的键即可。同时Spring Integration也提供了内置的错误处理机制来简化这个操作。

具体实现方案 #

1. 利用ExpressionEvaluatingRequestHandlerAdvice自动移除标记

可以通过配置请求处理器通知，在处理失败时自动从元数据存储中移除对应文件的标记：

@Bean
public ExpressionEvaluatingRequestHandlerAdvice sftpFailureAdvice(RedisMetadataStore metadataStore) {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 根据过滤器生成键的逻辑构造移除表达式，这里假设使用远程文件的绝对路径作为键
    advice.setOnFailureExpressionString(
        "metadataStore.remove(headers['file_remoteDirectory'] + '/' + headers['file_remoteFile'])");
    advice.setPropagateEvaluationFailures(false);
    return advice;
}

然后在处理文件的服务激活器上绑定这个通知：

@ServiceActivator(inputChannel = "sftpStreamingInputChannel", adviceChain = "sftpFailureAdvice")
public void processSftpFile(InputStream inputStream, 
                            @Header(FileHeaders.REMOTE_FILE) String remoteFileName,
                            @Header(FileHeaders.REMOTE_DIRECTORY) String remoteDir) {
    // 你的文件处理逻辑
}

2. 手动捕获异常并移除标记

如果需要更灵活的控制，可以在处理方法中捕获异常，手动调用RedisMetadataStore的remove方法：

@Autowired
private RedisMetadataStore metadataStore;

@Autowired
private SftpPersistentAcceptOnceFileListFilter sftpFilter;

@ServiceActivator(inputChannel = "sftpStreamingInputChannel")
public void processSftpFile(SftpFile remoteFile, InputStream inputStream) {
    try {
        // 文件处理逻辑
    } catch (Exception e) {
        // 生成和过滤器一致的键并移除
        String key = sftpFilter.generateKey(remoteFile);
        metadataStore.remove(key);
        // 重新抛出异常，确保重试机制生效
        throw new RuntimeException("文件处理失败，已移除标记以便重新拉取", e);
    }
}

注意：SftpPersistentAcceptOnceFileListFilter的generateKey方法默认返回文件的文件名（不含路径），如果你的SFTP监听目录包含子目录，可能会出现键冲突。这种情况下建议自定义过滤器重写generateKey方法，使用文件的远程绝对路径作为键：

public class UniqueKeySftpFilter extends SftpPersistentAcceptOnceFileListFilter {

    public UniqueKeySftpFilter(MetadataStore metadataStore, String prefix) {
        super(metadataStore, prefix);
    }

    @Override
    protected String generateKey(SftpFile file) {
        // 使用远程文件的绝对路径作为唯一键
        return file.getAbsolutePath();
    }
}

3. 事务同步优化方案（可选）

如果你希望从根源上避免“提前标记已处理”的问题，可以尝试将文件标记操作与处理逻辑绑定到同一事务中，只有处理成功才提交标记。不过在Streaming模式下，默认的文件标记是在读取文件时执行的，需要调整配置：

• 改用Pollable Inbound Channel Adapter配合FileSplitter，将文件内容拆分为消息流
• 配置事务管理器，确保过滤器的标记操作在事务提交后执行
• 处理失败时事务回滚，标记不会被写入Redis

这种方案适合对数据一致性要求较高的场景，但需要调整原有的Streaming模式架构。

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