Spring FTP入站通道适配器每隔一段时间记录一次FTP日志

Question

快速提问那么在FTP入站通道适配器中如何进行日志记录，例如每隔10分钟向远程FTP发送一次日志，轮询器是如何固定速率的呢?轮询器用于轮询，但它一直登录到远程服务器?

我有这个:

@Bean
    @InboundChannelAdapter(value = "stream", poller = @Poller(fixedRate = "1000"))
    public MessageSource<InputStream> ftpMessageSource() {
        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);
        messageSource.setRemoteDirectory(remotedirectory);
        messageSource.setFilter(filter());
        return messageSource;
    }

或轮询器元数据触发器:

 @Bean(name = PollerMetadata.DEFAULT_POLLER)
        public PollerMetadata defaultPoller() {
            PollerMetadata pollerMetadata = new PollerMetadata();
            pollerMetadata.setTrigger(new PeriodicTrigger(5000));
            return pollerMetadata;
        }

或者如何每10分钟记录一次日志，然后轮询所有新文件，设置Thread.sleep() ?

_______EDIT___

public static void main(String[] args) {
        SpringApplication.run(FtpinboundApp.class, args);
    }
    @Bean
    public SessionFactory<FTPFile> ftpSessionFactory() {
        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
        sf.setHost(remotehost);
        sf.setPort(remoteport);
        sf.setUsername(remoteuser);
        sf.setPassword(remotepassword);
        return new CachingSessionFactory<FTPFile>(sf);
    }
    @Bean
    @ServiceActivator(inputChannel = "data", adviceChain = "after")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                try {
                    httpposthgfiles.getHGFilesfromRestful(message.getPayload().toString());
                    httppost990.get990fromRestful(message.getPayload().toString());
                } catch (IOException e) {
                    logger.error(e);
                } catch (Exception e) {
                    logger.error(e);
                }
            }
        };
    }
    @Bean
    public ExpressionEvaluatingRequestHandlerAdvice after() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setOnSuccessExpression("@template.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
        advice.setPropagateEvaluationFailures(true);
        return advice;
    }
    @Bean
    @InboundChannelAdapter(value = "stream", poller = @Poller(fixedRate = "1000"))
    public MessageSource<InputStream> ftpMessageSource() {
        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);
        messageSource.setRemoteDirectory(remotedirectory);
        messageSource.setFilter(filter());
        return messageSource;
    }
    public FileListFilter<FTPFile> filter() {
        CompositeFileListFilter<FTPFile> filter = new CompositeFileListFilter<>();
        filter.addFilter(new FtpSimplePatternFileListFilter("xxxx_aaa204*"));
        filter.addFilter(acceptOnceFilter());
        return filter;
    }
    @Bean
    public FtpPersistentAcceptOnceFileListFilter acceptOnceFilter() {
        FtpPersistentAcceptOnceFileListFilter filter = new FtpPersistentAcceptOnceFileListFilter(meta(), "xxxx_aaa204");
        filter.setFlushOnUpdate(true);
        return filter;
    }
    @Bean
    public ConcurrentMetadataStore meta() {
        PropertiesPersistingMetadataStore meta = new PropertiesPersistingMetadataStore();
        meta.setBaseDirectory("/tmp/foo");
        meta.setFileName("ftpStream.properties");
        return meta;
    }
    @Bean
    @Transformer(inputChannel = "stream", outputChannel = "data")
    public org.springframework.integration.transformer.Transformer transformer() {
        return new StreamTransformer("UTF-8");
    }
    @Bean
    public FtpRemoteFileTemplate template() {
        return new FtpRemoteFileTemplate(ftpSessionFactory());
    }
    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata defaultPoller() {
        PollerMetadata pollerMetadata = new PollerMetadata();
        pollerMetadata.setTrigger(new PeriodicTrigger(5000));
        return pollerMetadata;
    }

Answer 1

只有在使用CachingSessionFactory时才会保持登录状态。

最好不要像那样休眠和绑定线程，而是使用任务调度器(这是轮询器所做的)。

new PeriodicTrigger(600_000)将每10分钟调度一个任务登录并检查文件。