停在 Service Activator 的 Spring 集成线程答案

【问题标题】：Spring Integration Threads Parked at Service Activator停在 Service Activator 的 Spring 集成线程
【发布时间】：2014-12-03 15:29:12
【问题描述】：

我遇到了线程停在我的服务激活器上的问题，这会导致在池耗尽时文件挂在 SftpGatewayChannel 中。我认为这与 SA 的返回无效有关，这是正确的，因为它们只是增加指标。

我能够通过向 SftpGateway 添加默认回复超时来解决此问题，但这并不理想，因为存在重试建议，并且如果存在连接问题，我不希望线程超时。我想要一个解决方案，在成功上传并调用“成功”服务激活器后将线程返回到池中。

<task:executor id="Tasker" rejection-policy="CALLER_RUNS" pool-size="${MaxThreads}" />

<int:channel id="SftpGatewayChannel">
    <int:dispatcher task-executor="Tasker" />
</int:channel>

<int:service-activator id="SegmentStart" input-channel="SftpGatewayChannel" ref="SftpGateway" />

<int:gateway id="SftpGateway" default-request-channel="SftpOutboundChannel" error-channel="ErrorChannel" />

<int:channel id="SftpOutboundChannel" datatype="java.lang.String,java.io.File,byte[]" />

<int-sftp:outbound-channel-adapter id="SftpOutboundAdapter"
    session-factory="SftpCachingSessionFactory" channel="SftpOutboundChannel" charset="UTF-8" >
    <int-sftp:request-handler-advice-chain>
        <ref bean="exponentialRetryAdvice" />
        <bean id="SuccessAdvice" class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice" >
            <property name="successChannel" ref="SuccessChannel"/>
            <property name="onSuccessExpression" value="true"/>
        </bean>
    </int-sftp:request-handler-advice-chain>
</int-sftp:outbound-channel-adapter>

<int:channel id="ErrorChannel">
    <int:interceptors>
        <int:wire-tap channel="FailureChannel" />
    </int:interceptors>
</int:channel>

<int:channel id="AttemptChannel" />
<int:channel id="SuccessChannel" />
<int:channel id="FailureChannel" />

<int:service-activator id="AttemptMetrics" input-channel="AttemptChannel"
    expression="T(MetricsCounter).addAttempt()"  />
<int:service-activator id="SuccessMetrics" input-channel="SuccessChannel"
    expression="T(MetricsCounter).addSuccesses(inputMessage.Headers.messages.size())" />
<int:service-activator id="FailureMetrics" input-channel="FailureChannel"
    expression="T(MetricsCounter).addFailures(payload.getFailedMessage().Headers.messages.size())" />

【问题讨论】：

标签： multithreading asynchronous spring-integration

【解决方案1】：

是的，网关默认会收到回复。除了使用默认的RequestReplyExchanger，您还可以使用service-interface 方法和void 返回void process(Message<?> m)。

或者，正如您所做的那样，只需在您的网关上添加default-reply-timeout="0"，线程就会立即返回，而无需等待回复（永远不会出现）。

...但这并不理想...

回复超时时钟只在线程返回网关时才开始，所以不会对下游流量造成影响。

【讨论】：