Spring Integration-停止入站适配器-观察到异常

问题描述

使用Spring Integration SFTP功能和入站通道适配器,可以下载远程文件进行处理。

虽然我使用以下启动和停止例程来启动/停止入站通道适配器。


    public void startInboundAdapter() throws ApplicationException {
        SourcePollingChannelAdapter streamInboundAdapter = appContext.getBean("streamInboundAdapter",SourcePollingChannelAdapter.class);
        streamInboundAdapter.start();
    }

    public void stopInboundAdapter() throws ApplicationException {
        SourcePollingChannelAdapter streamInboundAdapter = appContext.getBean("streamInboundAdapter",SourcePollingChannelAdapter.class);
        streamInboundAdapter.stop();
    }

但是在所有文件处理完成之后,调用stop inbound adapter之后观察到异常。我被困在这里,正在寻找解决方案,请提出建议。

Stacktrace


2020-08-15 15:44:24.298  INFO 9976 --- [   scheduling-1] o.s.i.e.sourcePollingChannelAdapter      : stopped bean 'streamInboundAdapter'
2020-08-15 15:44:24.298 ERROR 9976 --- [   scheduling-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [bean 'aggregatorfactorybean'; defined in: 'class path resource [com/config/SFTPConfig.class]'; from source: 'org.springframework.core.type.classreading.SimpleMethodMetadata@40dd3977']; nested exception is org.springframework.messaging.MessagingException: Interrupted while obtaining lock; nested exception is java.lang.InterruptedException,FailedMessage=Genericmessage [payload=sftp-inbound\upload_file.csv.bburpt.xls,headers={file_remoteHostPort=sftp.bloomberg.com:22,file_name=upload_file.csv.bburpt.xls,file_remoteDirectory=//report,file_originalFile=sftp-inbound\upload_file.csv.bburpt.xls,id=615015fe-b403-d247-c86e-f026a6ab3ad2,file_relativePath=upload_file.csv.bburpt.xls,file_remoteFile=upload_file.csv.bburpt.xls,timestamp=1597520532363}]
    at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:191)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:187)
    at org.springframework.integration.handler.ReplyProducingMessageHandlerWrapper.handleRequestMessage(ReplyProducingMessageHandlerWrapper.java:49)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:127)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:170)
    at org.springframework.integration.dispatcher.Abstractdispatcher.tryOptimizeddispatch(Abstractdispatcher.java:115)
    at org.springframework.integration.dispatcher.Unicastingdispatcher.dodispatch(Unicastingdispatcher.java:133)
    at org.springframework.integration.dispatcher.Unicastingdispatcher.dispatch(Unicastingdispatcher.java:106)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:403)
    at org.springframework.messaging.core.GenericmessagingTemplate.doSend(GenericmessagingTemplate.java:187)
    at org.springframework.messaging.core.GenericmessagingTemplate.doSend(GenericmessagingTemplate.java:166)
    at org.springframework.messaging.core.GenericmessagingTemplate.doSend(GenericmessagingTemplate.java:47)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
    at org.springframework.integration.endpoint.sourcePollingChannelAdapter.handleMessage(SourcePollingChannelAdapter.java:234)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.messageReceived(AbstractPollingEndpoint.java:396)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:380)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:328)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:275)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$2(AbstractPollingEndpoint.java:272)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.messaging.MessagingException: Interrupted while obtaining lock; nested exception is java.lang.InterruptedException
    at org.springframework.integration.store.SimpleMessageStore.completeGroup(SimpleMessageStore.java:426)
    at org.springframework.integration.aggregator.AggregatingMessageHandler.afterRelease(AggregatingMessageHandler.java:80)
    at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.processMessageForGroup(AbstractCorrelatingMessageHandler.java:503)
    at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.handleMessageInternal(AbstractCorrelatingMe ssageHandler.java:471)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:170)
    ... 31 more
Caused by: java.lang.InterruptedException
    at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1261)
    at java.base/java.util.concurrent.locks.reentrantlock.lockInterruptibly(reentrantlock.java:317)
    at org.springframework.integration.store.SimpleMessageStore.completeGroup(SimpleMessageStore.java:412)
    ... 35 more

解决方法

停止是立即的,如果您有一个可中断的正在进行的任务,它将被中断。

请添加新的功能建议issue on GitHub

我们应该实施Pausable,让您等到它暂停后再停止,或者为停止添加“有序”选项。