在@InboundChannelAdapter场景下，发送AWS SQS消息后如何归档文件？

要实现AWS SQS消息发送成功后将文件移动至远程归档目录的需求，你可以沿用事务同步的思路，或使用Spring Integration的请求处理器通知来绑定发送成功后的操作，以下是具体实现：

方式一：基于事务同步扩展（适配SQS场景） #

尽管AWS SQS本身不支持事务，但你可以将SQS发送操作与本地事务同步绑定，确保发送完成后触发归档逻辑：

1. 配置事务同步处理器与工厂

@Bean
public TransactionSynchronizationFactory sqsTransactionSynchronizationFactory() {
    ExpressionEvaluatingTransactionSynchronizationProcessor processor = 
        new ExpressionEvaluatingTransactionSynchronizationProcessor();
    // 替换为你的归档逻辑表达式，假设消息头中存储了文件路径
    processor.setAfterCommitExpression(
        PARSER.parseExpression("T(com.yourpackage.FileArchiver).archiveFile(headers['filePath'])")
    );
    return new DefaultTransactionSynchronizationFactory(processor);
}

2. 为SqsMessageHandler添加事务同步拦截器
   修改你的sqsMessageHandler Bean，注入事务同步工厂并配置拦截器：

@Bean
@ServiceActivator(inputChannel = "sqsRequisitionToAlmSendChannel")
public MessageHandler sqsMessageHandler(@Autowired SqsAsyncClient amazonSqs,
                                        @Autowired TransactionSynchronizationFactory sqsTransactionSynchronizationFactory) {
    SqsMessageHandler sqsMessageHandler = new SqsMessageHandler(amazonSqs);
    sqsMessageHandler.setQueueNotFoundStrategy(QueueNotFoundStrategy.FAIL);
    sqsMessageHandler.setQueue(queue_name);
    
    TransactionSynchronizationInterceptor interceptor = 
        new TransactionSynchronizationInterceptor(sqsTransactionSynchronizationFactory);
    sqsMessageHandler.setAdviceChain(Collections.singletonList(interceptor));
    
    return sqsMessageHandler;
}

方式二：使用请求处理器通知（直接绑定成功回调） #

如果无需事务绑定，可直接通过ExpressionEvaluatingRequestHandlerAdvice实现发送成功后的归档操作：

1. 配置成功回调通知

@Bean
public ExpressionEvaluatingRequestHandlerAdvice sqsSuccessAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 发送成功后执行的归档逻辑表达式
    advice.setOnSuccessExpression(
        PARSER.parseExpression("T(com.yourpackage.FileArchiver).archiveFile(headers['filePath'])")
    );
    // 可选：配置发送失败时的处理逻辑
    advice.setOnFailureExpression(PARSER.parseExpression("T(com.yourpackage.FileArchiver).handleFailure(headers['filePath'], exception)"));
    advice.setTrapException(false); // 不捕获异常，保留异常传播逻辑
    return advice;
}

2. 为SqsMessageHandler添加通知

@Bean
@ServiceActivator(inputChannel = "sqsRequisitionToAlmSendChannel")
public MessageHandler sqsMessageHandler(@Autowired SqsAsyncClient amazonSqs,
                                        @Autowired ExpressionEvaluatingRequestHandlerAdvice sqsSuccessAdvice) {
    SqsMessageHandler sqsMessageHandler = new SqsMessageHandler(amazonSqs);
    sqsMessageHandler.setQueueNotFoundStrategy(QueueNotFoundStrategy.FAIL);
    sqsMessageHandler.setQueue(queue_name);
    
    sqsMessageHandler.setAdviceChain(Collections.singletonList(sqsSuccessAdvice));
    
    return sqsMessageHandler;
}

关键注意事项 #

• 确保消息中携带文件路径信息（可放在消息头headers['filePath']或Payload中），供归档逻辑获取目标文件
• 若使用异步SQS客户端，上述两种方式都会在SQS发送确认完成后触发归档操作
• 需自行实现FileArchiver类中的远程目录文件移动逻辑（如基于SFTP、S3等客户端）

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