java - Spring 集成 : individual timeouts settings for connections in the same pool

Question

我们有一个使用 Spring Integration 的 Java 编写的应用程序。应用程序向 3d 方服务发送请求，每个请求都表示为字节数组并通过纯 TCP 发送。与 3d 方的连接存储在池中(使用 CachingClientConnectionFactory)。

我们可以向 3d 方发送几种类型的请求，现在需要为每种类型的请求设置不同的超时值。但是，就目前而言，这看起来有问题，因为为 Gateway 和 ConnectionFactory 组件设置了连接超时设置，并且无法为特定的单个请求设置连接超时。

我们希望避免仅仅为了支持不同的连接超时而引入多个网关和连接工厂。

channel 和网关配置

<int:channel id="myInput" />

<int:gateway id="myGateway" 
    service-interface="com.mypackage.TcpGateway"
    default-request-channel="myInput"/>

<int-ip:tcp-outbound-gateway id="myOutGateway"
    request-channel="mybInput"
    reply-channel="clientBytesChannel"
    connection-factory="myConnectionFactory"
    request-timeout="${conn.timeout}"
    remote-timeout="${conn.timeout}"/>

TcpGateway.java

package com.mypackage.TcpGateway;

public interface TcpGateway {
    byte[] send(byte[] message);
}

RequestProcessor.java

public class RequestProcessors {
    @Autowired
    private TcpGateway myGateway;

    public MyResponse process(MyRequest requestMessage) {
        byte[] binaryMessage = transformRequest(requestMessage);
        byte[] response = myGateway.send(binaryMessage);
        return transformResponse(response);
    }

    // rest of business logic here
}

查看上面的源代码，似乎最简单的方法是扩展原生组件 TcpOutboundGateway 并替换属性 remoteTimeout使用将超时设置保存为 ThreadLocal 的单例 bean多变的。然后可以根据 RequestProcessors.process() 中的请求类型设置必要的值。请求传递给 myGateway.send() 之前的方法.

但是，我无法找到一种优雅的方式来使用我的自定义类重新定义 TcpOutboundGateway 组件。分析了Spring Integration的源码，似乎TcpOutboundGateway对输入请求一无所知，而是注册在一个消息处理器链中，需要时调用。所以，现在这看起来不像是一个简单的解决方案。

如果您对如何更改标签 <int-ip:tcp-outbound-gateway> 使用的类有任何想法，或者如果您对如何以完全不同的方式解决主要超时问题有任何想法，请提出建议。

谢谢。

Answer 1

有趣的问题。

您是正确的(当前)AsyncReply对象不知道原始出站消息。

我们可以做的一件事是添加一个 remote-timeout-expression并根据消息计算超时...

AsyncReply reply = new AsyncReply(calculateReplyTimeout(requestMessage));

随意开一个JIRA Improvement Issue .

同时，如果您想自定义网关，您可以轻松实现。

虽然命名空间 ( <int-ip:... ) 不支持自定义类，但您始终可以使用 <bean/> 连接网关

你需要一个ConsumerEndpointFactoryBean并在其 handler 中获取您的自定义网关属性(property)。