FTP 入站通道适配器在连接重置后停止工作

Question

我们目前正在使用 Spring Integration 3.0.2 轮询 FTP 文件夹中的 XLSX 文件。但是，重置数据连接时，轮询器将停止工作。FTP 服务器上的所有文件都保留在 FTP 上，因此不执行轮询。

我已经读到在 SI 的 v 3.0 之前，连接被缓存并导致一些问题，但通常情况并非如此，因为我们使用普通的 FtpSessionFactory。

堆栈跟踪：

2015-01-06 16:26:15,444 ERROR [] () [de.hybris.platform.acceleratorservices.dataimport.batch.task.ErrorHandler] unexpected exception caught
org.springframework.integration.MessagingException: Problem occurred while synchronizing remote to local directory
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:189)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource.receive(AbstractInboundFileSynchronizingMessageSource.java:162)
    at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:111)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:184)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.access$000(AbstractPollingEndpoint.java:51)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:143)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:141)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:273)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:52)
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:49)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:49)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:268)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:53)
    at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    at java.util.concurrent.FutureTask.run(FutureTask.java:262)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:178)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:292)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.integration.MessagingException: Failed to execute on session
    at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:306)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:163)
    ... 20 more
Caused by: java.lang.IllegalStateException: failed to create FTPClient
    at org.springframework.integration.ftp.session.AbstractFtpSessionFactory.getSession(AbstractFtpSessionFactory.java:139)
    at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:295)
    ... 21 more
Caused by: java.net.SocketException: Connection reset
    at java.net.SocketInputStream.read(SocketInputStream.java:196)
    at java.net.SocketInputStream.read(SocketInputStream.java:122)
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283)
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325)
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177)
    at java.io.InputStreamReader.read(InputStreamReader.java:184)
    at java.io.BufferedReader.fill(BufferedReader.java:154)
    at java.io.BufferedReader.read(BufferedReader.java:175)
    at org.apache.commons.net.io.CRLFLineReader.readLine(CRLFLineReader.java:58)
    at org.apache.commons.net.ftp.FTP.__getReply(FTP.java:310)
    at org.apache.commons.net.ftp.FTP.__getReply(FTP.java:290)
    at org.apache.commons.net.ftp.FTP._connectAction_(FTP.java:392)
    at org.apache.commons.net.ftp.FTPClient._connectAction_(FTPClient.java:764)
    at org.apache.commons.net.SocketClient.connect(SocketClient.java:169)
    at org.apache.commons.net.SocketClient.connect(SocketClient.java:189)
    at org.springframework.integration.ftp.session.AbstractFtpSessionFactory.createClient(AbstractFtpSessionFactory.java:152)
    at org.springframework.integration.ftp.session.AbstractFtpSessionFactory.getSession(AbstractFtpSessionFactory.java:136)
    ... 22 more

配置：

<int:channel id="ftpChannel">
    <int:queue/>
</int:channel>
<bean id="ftpSessionFactory"
    class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
    <property name="host" value="abcd.com"/>
    <property name="port" value="21"/>
    <property name="username" value="username"/>
    <property name="password" value="password"/>
    <property name="fileType" value="2"/>
    <property name="clientMode" value="2" />
</bean>
<int-ftp:inbound-channel-adapter id="supplierImportFtpChannel"
                                 channel="ftpChannel"
                                 session-factory="ftpSessionFactory"
                                 local-directory="/localdirectory"
                                 remote-directory="/ftpdirectory"
                                 filename-regex=".*.xlsx$"
                                 delete-remote-files="true"
                                 temporary-file-suffix=".writing">
    <int:poller fixed-rate="1000"/>
</int-ftp:inbound-channel-adapter>

我错过了什么吗？

Answer 1

框架中没有任何内容可以在此类异常后"停止"轮询器。

最有可能的是，轮询器线程在尝试连接到服务器的下一次轮询中挂起 - 您可以使用线程转储（使用 jstack）进行验证。

连接时没有超时（默认情况下）。

从 4.0 开始，您现在可以直接在 DefaultFtpSessionFactory 上配置超时;在 4.0 之前，您必须使用文档中描述的技术（向下滚动到高级配置）。