当 rabbitmq 关闭时停止轮询文件：spring 集成

Question

我正在做一个项目，我们正在从 sftp 服务器轮询文件并将其流式传输到 rabbitmq 队列上的对象中。现在，当 rabbitmq 关闭时，它仍然会从服务器轮询并删除文件，并在 rabbitmq 关闭时将文件发送到队列中时丢失文件。我正在使用 ExpressionEvaluatingRequestHandlerAdvice 删除成功转换的文件。我的代码如下所示：

@Bean
public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() {
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
    factory.setHost(sftpProperties.getSftpHost());
    factory.setPort(sftpProperties.getSftpPort());
    factory.setUser(sftpProperties.getSftpPathUser());
    factory.setPassword(sftpProperties.getSftpPathPassword());
    factory.setAllowUnknownKeys(true);
    return new CachingSessionFactory<>(factory);
}    

@Bean
public SftpRemoteFileTemplate sftpRemoteFileTemplate() {
    return new SftpRemoteFileTemplate(sftpSessionFactory());
}

@Bean
@InboundChannelAdapter(channel = TransformerChannel.TRANSFORMER_OUTPUT, autoStartup = "false",
        poller = @Poller(value = "customPoller"))
public MessageSource<InputStream> sftpMessageSource() {
    SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(sftpRemoteFileTemplate,
            null);
    messageSource.setRemoteDirectory(sftpProperties.getSftpDirPath());
    messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),
            "streaming"));
    messageSource.setFilter(new SftpSimplePatternFileListFilter("*.txt"));
    return messageSource;
}    


@Bean
@Transformer(inputChannel = TransformerChannel.TRANSFORMER_OUTPUT,
        outputChannel = SFTPOutputChannel.SFTP_OUTPUT,
        adviceChain = "deleteAdvice")
public org.springframework.integration.transformer.Transformer transformer() {
    return new SFTPTransformerService("UTF-8");
}

@Bean
public ExpressionEvaluatingRequestHandlerAdvice deleteAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setOnSuccessExpressionString(
                "@sftpRemoteFileTemplate.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
    advice.setPropagateEvaluationFailures(false);
    return advice;
}

当 rabbitmq 服务器关闭时，我不希望文件从远程 sftp 服务器中删除/轮询。我怎样才能做到这一点？

更新

抱歉，我没有提到我正在使用 Spring Cloud Stream 兔子活页夹。这是变压器服务：

public class SFTPTransformerService extends StreamTransformer {
public SFTPTransformerService(String charset) {
    super(charset);
}

@Override
protected Object doTransform(Message<?> message) throws Exception {
    String fileName = message.getHeaders().get("file_remoteFile", String.class);
    Object fileContents = super.doTransform(message);
    return new customFileDTO(fileName, (String) fileContents);
}
}

UPDATE-2 我按照建议在 customPoller 上添加了TransactionSynchronizationFactory。现在当rabbit服务器关闭时它不会轮询文件，但是当服务器启动时，它会一遍又一遍地轮询同一个文件！我想不通为什么？我想我不能使用PollerSpec，因为我在 4.3.2 版本上。

@Bean(name = "customPoller")
public PollerMetadata pollerMetadataDTX(StartStopTrigger startStopTrigger,
                                        CustomTriggerAdvice customTriggerAdvice) {
    PollerMetadata pollerMetadata = new PollerMetadata();
    pollerMetadata.setAdviceChain(Collections.singletonList(customTriggerAdvice));
    pollerMetadata.setTrigger(startStopTrigger);
    pollerMetadata.setMaxMessagesPerPoll(Long.valueOf(sftpProperties.getMaxMessagePoll()));
    ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
            new ExpressionEvaluatingTransactionSynchronizationProcessor();
    syncProcessor.setBeanFactory(applicationContext.getAutowireCapableBeanFactory());
    syncProcessor.setBeforeCommitChannel(
            applicationContext.getBean(TransformerChannel.TRANSFORMER_OUTPUT, MessageChannel.class));
    syncProcessor
            .setAfterCommitChannel(
                    applicationContext.getBean(SFTPOutputChannel.SFTP_OUTPUT, MessageChannel.class));
    syncProcessor.setAfterCommitExpression(new SpelExpressionParser().parseExpression(
            "@sftpRemoteFileTemplate.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])"));
    DefaultTransactionSynchronizationFactory defaultTransactionSynchronizationFactory =
            new DefaultTransactionSynchronizationFactory(syncProcessor);
    pollerMetadata.setTransactionSynchronizationFactory(defaultTransactionSynchronizationFactory);
    return pollerMetadata;
}

我不知道您是否需要此信息，但我的 CustomTriggerAdvice 和 StartStopTrigger 看起来像这样：

@Component
public class CustomTriggerAdvice extends AbstractMessageSourceAdvice {

@Autowired private StartStopTrigger startStopTrigger;

@Override
public boolean beforeReceive(MessageSource<?> source) {
    return true;
}

@Override
public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
    if (result == null) {
        if (startStopTrigger.getStart()) {
            startStopTrigger.stop();
        }
    } else {
        if (!startStopTrigger.getStart()) {
            startStopTrigger.stop();
        }
    }
    return result;
}

}


public class StartStopTrigger implements Trigger {

private PeriodicTrigger startTrigger;
private boolean start;

public StartStopTrigger(PeriodicTrigger startTrigger, boolean start) {
    this.startTrigger = startTrigger;
    this.start = start;
}

@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
    if (!start) {
        return null;
    }
    start = true;
    return startTrigger.nextExecutionTime(triggerContext);
}

public void stop() {
    start = false;
}

public void start() {
    start = true;
}

public boolean getStart() {
    return this.start;
}
}

Answer 1

好吧，很高兴看看您的SFTPTransformerService 是什么，并确定在出现异常情况时如何执行onSuccessExpression，以防代理失败。

您还应该不仅抛出异常不执行删除，而是考虑添加RequestHandlerRetryAdvice以重新发送文件到RabbitMQ：https://docs.spring.io/spring-integration/docs/5.0.6.RELEASE/reference/html/messaging-endpoints-chapter.html#retry-advice

更新

所以，好吧，既然 Gary 猜到您在内部处理之后使用 Spring Cloud Stream 向 Rabbit Binder 发送消息（很遗憾您最初没有共享该信息），您需要查看 Binder关于此事的错误处理：https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/reference/htmlsingle/#_retry_with_the_rabbitmq_binder

确实ExpressionEvaluatingRequestHandlerAdvice 仅适用于SFTPTransformerService，仅此而已。下游错误（在 Binder 中）尚未包含在此过程中。

更新 2

是的...我认为 Gary 是对的，我们别无选择，除非在 customPoller 级别上配置 TransactionSynchronizationFactory 而不是 ExpressionEvaluatingRequestHandlerAdvice：ExpressionEvaluatingRequestHandlerAdvice。

DefaultTransactionSynchronizationFactory 可以使用ExpressionEvaluatingTransactionSynchronizationProcessor 进行配置，其目标与提到的ExpressionEvaluatingRequestHandlerAdvice 相似，但在事务级别上，它将包括您的进程，从 SFTP 通道适配器开始，到 Rabbit Binder 级别结束与发送到 AMQP 的尝试。

有关详细信息，请参阅参考手册：https://docs.spring.io/spring-integration/reference/html/transactions.html#transaction-synchronization。

ExpressionEvaluatingRequestHandlerAdvice（和任何AbstractRequestHandlerAdvice）的要点是它们仅在handleRequestMessage() 方法周围有边界，因此仅在组件期间声明它们。