从Unix读取而不使用Spring Integration复制到本地时发现的循环引用

2023年8月15日 201次阅读

我想从Unix位置读取.txt文件而不使用
Spring Integration将其复制到本地.这应该以连续模式完成,即当新文件到来时应该检测和读取.

码：

 @SpringBootApplication
    public class SftpJavaApplication {

        public static void main(String[] args) {
            new SpringApplicationBuilder(SftpJavaApplication.class)
                .web(false)
                .run(args);
        }
        @Bean
        public SessionFactory<LsEntry> sftpSessionFactory() {

            DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
            factory.setHost("ip");
            factory.setPort(port);
            factory.setUser("user");
            factory.setPassword("pwd");
            factory.setAllowUnknownKeys(true);
            return new CachingSessionFactory<LsEntry>(factory);
        }
      @Bean
@Transformer(inputChannel = "stream",outputChannel="data")
public org.springframework.integration.transformer.Transformer  transformer () {
    return  new org.springframework.integration.transformer.StreamTransformer("UTF-8");
            }

@Bean
@InboundChannelAdapter(value = "stream", poller = @Poller(fixedDelay = "1000", maxMessagesPerPoll = "1"))
public MessageSource<InputStream> ftpMessageSource() {
    SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null);
    messageSource.setRemoteDirectory("/test1/test2/test3");
    messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),
                       "streaming"));
    return messageSource;
}



@Bean
public SftpRemoteFileTemplate template() {
    return new SftpRemoteFileTemplate(sftpSessionFactory());

}

@Bean
@ServiceActivator(inputChannel = "data" )
public MessageHandler handler() {
    return new MessageHandler() {


        @Override
        public void handleMessage(Message<?> message) throws MessagingException {
            System.out.println(">>>>>>>>>>>>>"+message.getPayload()); //this prints the data in the file


        }

    };
}

}

依赖性：

 compile("org.springframework.cloud:spring-cloud-spring-service-connector:1.2.1.RELEASE")
    compile("org.springframework.cloud:spring-cloud-cloudfoundry-connector:1.2.1.RELEASE")
    compile("org.springframework.boot:spring-boot-starter-integration")
    compile group: 'com.jcraft', name: 'jsch', version: '0.1.44-1'
    compile group: 'org.springframework.integration', name: 'spring-integration-sftp', version: '4.3.1.RELEASE'
    compile group: 'org.springframework.integration', name: 'spring-integration-file', version: '4.3.1.RELEASE'
    compile('org.kie.modules:org-apache-commons-lang3:6.3.0.Final')
    compile("com.h2database:h2:1.4.192")

编译组：’org.springframework.integration’,名称：’spring-integration-core’,版本：’4.3.1.RELEASE’

堆栈跟踪 ：

Caused by: org.springframework.core.NestedIOException: Failed to list files; nested exception is 2: No such file
    at org.springframework.integration.sftp.session.SftpSession.list(SftpSession.java:104)
    at org.springframework.integration.sftp.session.SftpSession.list(SftpSession.java:50)
    at org.springframework.integration.file.remote.session.CachingSessionFactory$CachedSession.list(CachingSessionFactory.java:218)
    at org.springframework.integration.file.remote.RemoteFileTemplate$6.doInSession(RemoteFileTemplate.java:417)
    at org.springframework.integration.file.remote.RemoteFileTemplate$6.doInSession(RemoteFileTemplate.java:413)
    at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:435)
    ... 24 more
Caused by: 2: No such file
    at com.jcraft.jsch.ChannelSftp.throwStatusError(ChannelSftp.java:2297)
    at com.jcraft.jsch.ChannelSftp._stat(ChannelSftp.java:1750)
    at com.jcraft.jsch.ChannelSftp._stat(ChannelSftp.java:1767)
    at com.jcraft.jsch.ChannelSftp.ls(ChannelSftp.java:1205)
    at org.springframework.integration.sftp.session.SftpSession.list(SftpSession.java:92)
    ... 29 more

最佳答案 从Spring Integration 4.3开始,远程文件支持(FTP / SFTP)提供流式适配器：

<int-ftp:inbound-streaming-channel-adapter id="ftpInbound"
            channel="ftpChannel"
            session-factory="sessionFactory"
            filename-pattern="*.txt"
            filename-regex=".*\.txt"
            filter="filter"
            remote-file-separator="/"
            comparator="comparator"
            remote-directory-expression="'foo/bar'">
        <int:poller fixed-rate="1000" />
</int-ftp:inbound-streaming-channel-adapter>

http://docs.spring.io/spring-integration/reference/html/whats-new.html#_ftp_sftp_streaming_inbound_channel_adapters

但！我们不能“以连续模式”这样做,因为(S)FTP不是事件驱动的协议.所以,我们仍然应该定期轮询远程目录.

如果你真的知道一些方法让它在远程目录中监听一些事件,我们很高兴在Spring Integration中开发这样一个组件.

编辑

当您升级Spring Integration的依赖项时,请确保所有模块都在同一版本中,您的版本控制并不好.

spring-integration-file也应该是4.3.1.RELEASE.

OTOH你根本不需要它.它是spring-integration-sftp的传递依赖.
除了你不需要spring-integration-core之外,因为它是所有这些的传递依赖.

EDIT2

I would only like to know the significance of the word “Streaming ” used in the filter .It as such did not prevent file from reading at my end. But for knowledge purpose i would like to know

Streaming适配器的目的是让目标应用程序不创建本地文件副本.只需直接从内存中读取远程数据源中的数据即可.所以,这就是你仍然可以读取文件的方式,但这是一个遥远的文件.不是吗？

关于PCF的错误.

与我们分享关于此问题的StackTrace会很棒.
从另一方面来看,请尝试对DefaultSftpSessionFactory使用allowUnknownKeys = true.并在Reference Manual阅读更多内容.

点赞