原因：com.jcraft.jsch.JSchException：使用 DefaultSftpSessionFactory 和 SftpInboundFileSynchronizer 时连接被外部主机关闭

Question

我正在使用 Spring Batch 远程分区来读取输入文件并处理它们。为了使输入文件在所有服务器上可用，我添加了步骤侦听器，它检查是否存在并从主服务器下载它们。

<step id="importExchangesStep">
    <tasklet transaction-manager="transactionManager">
        <chunk reader="importExchangesFileItemReader" writer="importExchangesItemWriter"
            commit-interval="${import.exchanges.commit.interval}" />
        <listeners>
            <listener ref="ftpGetRemoteExchangesFilesListener" />
        </listeners>
    </tasklet>
</step>

<job id="importExchangesJob" restartable="true">
    <step id="importExchangesStep.master">
        <partition partitioner="importExchangesPartitioner"
            handler="importExchangesPartitionHandler" />
    </step>
</job>

我正在使用 DefaultSftpSessionFactory 将文件从主服务器下载到从服务器。有 4 台服务器，每台服务器的消费者并发度为 7，因此总共有 28 个分区（步骤执行）并行运行。

以下是sftp配置，

<beans:bean id="sftpSessionFactory"
    class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
    <beans:property name="host" value="${master.host}" />
    <beans:property name="user" value="${master.user}" />
    <beans:property name="password" value="${master.password}" />
    <beans:property name="port" value="22"/>
</beans:bean>

如果只在一台服务器上运行它就可以工作。但如果我将其分发到 4 个某些分区成功完成，而某些分区则因此异常而失败

Caused by: com.jcraft.jsch.JSchException: connection is closed by foreign host

监听器配置：

<beans:bean id="ftpGetRemoteExchangesFilesListener"
    class="com.st.batch.listeners.FtpGetRemoteFilesListener"
    p:sessionFactory-ref="sftpSessionFactory" p:downloadFileAttempts="3"
    p:fileNamePattern="*.txt" p:deleteLocalFiles="false"
    p:localDirectory="/tmp/spring/batch/#{jobParameters[batch_id]}/exchanges/"
    p:remoteDirectory="/tmp/spring/batch/#{jobParameters[batch_id]}/exchanges/"
    scope="step" />

没有引用此的 getter 和 setter 的监听器类

http://coreyreil.wordpress.com/2012/12/21/spring-batch-creating-an-ftp-tasklet-to-get-remote-files/

public class FtpGetRemoteFilesListener extends StepExecutionListenerSupport implements InitializingBean
{
    //private Logger logger = LoggerFactory.getLogger(FtpGetRemoteFilesTasklet.class);
    private static Log logger = LogFactory.getLog(FtpGetRemoteFilesListener.class);

    private File localDirectory;

    private AbstractInboundFileSynchronizer<?> ftpInboundFileSynchronizer;

    private SessionFactory sessionFactory;

    private boolean autoCreateLocalDirectory = true;

    private boolean deleteLocalFiles = true;

    private String fileNamePattern;

    private String remoteDirectory;

    private int downloadFileAttempts = 12;

    private long retryIntervalMilliseconds = 300000;

    private boolean retryIfNotFound = false;


    /* (non-Javadoc)
     * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
     */
    public void afterPropertiesSet() throws Exception
    {
        Assert.notNull(sessionFactory, "sessionFactory attribute cannot be null");
        Assert.notNull(localDirectory, "localDirectory attribute cannot be null");
        Assert.notNull(remoteDirectory, "remoteDirectory attribute cannot be null");
        Assert.notNull(fileNamePattern, "fileNamePattern attribute cannot be null");

        setupFileSynchronizer();

        if (!this.localDirectory.exists())
        {
            if (this.autoCreateLocalDirectory)
            {
                if (logger.isDebugEnabled())
                {
                    logger.debug("The '" + this.localDirectory + "' directory doesn't exist; Will create.");
                }
                this.localDirectory.mkdirs();
            }
            else
            {
                throw new FileNotFoundException(this.localDirectory.getName());
            }
        }
    }

    private void setupFileSynchronizer()
    {
        if (isSftp())
        {
            ftpInboundFileSynchronizer = new SftpInboundFileSynchronizer(sessionFactory);
            ((SftpInboundFileSynchronizer) ftpInboundFileSynchronizer).setFilter(new SftpSimplePatternFileListFilter(fileNamePattern));
        }
        else
        {
            ftpInboundFileSynchronizer = new FtpInboundFileSynchronizer(sessionFactory);
            ((FtpInboundFileSynchronizer) ftpInboundFileSynchronizer).setFilter(new FtpSimplePatternFileListFilter(fileNamePattern));
        }
        ftpInboundFileSynchronizer.setRemoteDirectory(remoteDirectory);
    }

    private void deleteLocalFiles()
    {
        if (deleteLocalFiles)
        {
            SimplePatternFileListFilter filter = new SimplePatternFileListFilter(fileNamePattern);
            List<File> matchingFiles = filter.filterFiles(localDirectory.listFiles());
            if (CollectionUtils.isNotEmpty(matchingFiles))
            {
                for (File file : matchingFiles)
                {
                    FileUtils.deleteQuietly(file);
                }
            }
        }
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {

        deleteLocalFiles();

        ftpInboundFileSynchronizer.synchronizeToLocalDirectory(localDirectory);

        if (retryIfNotFound)
        {
            SimplePatternFileListFilter filter = new SimplePatternFileListFilter(fileNamePattern);
            int attemptCount = 1;
            while (filter.filterFiles(localDirectory.listFiles()).size() == 0 && attemptCount <= downloadFileAttempts)
            {
                logger.info("File(s) matching " + fileNamePattern + " not found on remote site.  Attempt " + attemptCount + " out of " + downloadFileAttempts);
                try {
                    Thread.sleep(retryIntervalMilliseconds);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                ftpInboundFileSynchronizer.synchronizeToLocalDirectory(localDirectory);
                attemptCount++;
            }

            if (attemptCount >= downloadFileAttempts && filter.filterFiles(localDirectory.listFiles()).size() == 0)
            {
                try {
                    throw new FileNotFoundException("Could not find remote file(s) matching " + fileNamePattern + " after " + downloadFileAttempts + " attempts.");
                } catch (FileNotFoundException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }

    }
}

日志：

12:28:47,430 ERROR SimpleAsyncTaskExecutor-3 step.AbstractStep:225 - Encountered an error executing step importExchangesStep in job importExchangesJob
java.lang.IllegalStateException: failed to create SFTP Session
    at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:266)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:143)
    at com.st.batch.listeners.FtpGetRemoteFilesListener.beforeStep(FtpGetRemoteFilesListener.java:121)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:317)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:183)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:150)
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:132)
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:120)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)
    at $Proxy29.beforeStep(Unknown Source)
    at org.springframework.batch.core.listener.CompositeStepExecutionListener.beforeStep(CompositeStepExecutionListener.java:77)
    at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:194)
    at org.springframework.batch.integration.partition.StepExecutionRequestHandler.handle(StepExecutionRequestHandler.java:64)
    at sun.reflect.GeneratedMethodAccessor121.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at org.springframework.expression.spel.support.ReflectiveMethodExecutor.execute(ReflectiveMethodExecutor.java:69)
    at org.springframework.expression.spel.ast.MethodReference$MethodValueRef.getValue(MethodReference.java:97)
    at org.springframework.expression.spel.ast.CompoundExpression.getValueInternal(CompoundExpression.java:81)
    at org.springframework.expression.spel.ast.SpelNodeImpl.getTypedValue(SpelNodeImpl.java:102)
    at org.springframework.expression.spel.standard.SpelExpression.getValue(SpelExpression.java:103)
    at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:126)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:227)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:127)
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:73)
    at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:67)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:134)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:115)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:102)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)
    at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)
    at org.springframework.integration.core.MessagingTemplate.doSendAndReceive(MessagingTemplate.java:318)
    at org.springframework.integration.core.MessagingTemplate.sendAndReceive(MessagingTemplate.java:239)
    at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:233)
    at org.springframework.integration.gateway.MessagingGatewaySupport.sendAndReceiveMessage(MessagingGatewaySupport.java:207)
    at org.springframework.integration.amqp.inbound.AmqpInboundGateway.access$200(AmqpInboundGateway.java:47)
    at org.springframework.integration.amqp.inbound.AmqpInboundGateway$1.onMessage(AmqpInboundGateway.java:87)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:693)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:586)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:75)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:154)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1113)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:559)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:904)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:888)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$500(SimpleMessageListenerContainer.java:75)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:989)
    at java.lang.Thread.run(Unknown Source)
Caused by: java.lang.IllegalStateException: failed to connect
    at org.springframework.integration.sftp.session.SftpSession.connect(SftpSession.java:204)
    at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:262)
    ... 54 more
Caused by: com.jcraft.jsch.JSchException: connection is closed by foreign host
    at com.jcraft.jsch.Session.connect(Session.java:244)
    at com.jcraft.jsch.Session.connect(Session.java:158)
    at org.springframework.integration.sftp.session.SftpSession.connect(SftpSession.java:196)
    ... 55 more

同时连接的数量是否有限制，因为必须有 28 个分区尝试连接，其中一些在所有服务器上成功，一些失败，或者其他原因？

我可以使用 sftp user@host 通过命令行从所有服务器登录到 master。

Answer 1

听起来您的 SSH 服务器对来自服务器的并发连接数有一些限制。看起来您正在使用 Spring Integration 2.2.x（由堆栈跟踪表明）。

Spring Integration 2.2.x 对每个会话使用一个连接。

3.0 引入了共享会话的概念，其中每个“会话”是通过单个共享连接/会话复用的通道。

添加

<constructor-arg value="true"/>

以启用此功能。

如果无法升级到 3.0，则必须考虑配置服务器以允许更多连接。