如何在spring集成TCP服务器中等待数据

Question

我使用Spring集成构建的TCP服务器工作得很好。我使用ByteArrayLengthHeaderSerializer作为序列化程序。

偶尔，客户端数据来的非常慢，使服务器的响应非常慢。

我想等待最多5秒从客户端读取数据的每个字节。如果数据字节没有在5秒内出现，我想发送NAK。

如何将超时设置为5秒？应该在哪里设置呢？是否需要自定义序列化程序？

以下是我的春季背景：

<int-ip:tcp-connection-factory id="crLfServer"
        type="server"
        port="${availableServerSocket}"
        single-use="true"
        so-timeout="10000"
        using-nio="false" 
        serializer="connectionSerializeDeserialize"
        deserializer="connectionSerializeDeserialize"
        so-linger="2000"/>

    <bean id="connectionSerializeDeserialize" class="org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer"/>

    <int-ip:tcp-inbound-gateway id="gatewayCrLf"
        connection-factory="crLfServer"
        request-channel="serverBytes2StringChannel"
        error-channel="errorChannel"
        reply-timeout="10000"/> <!-- reply-timeout works on inbound-gateway -->

    <int:channel id="toSA" />

    <int:service-activator input-channel="toSA"
        ref="myService"
        method="prepare"/>

    <int:object-to-string-transformer id="serverBytes2String"
        input-channel="serverBytes2StringChannel"
        output-channel="toSA"/>

    <int:transformer id="errorHandler"
        input-channel="errorChannel"
        expression="payload.failedMessage.payload + ':' + payload.cause.message"/>

谢谢

Answer 1

您需要一个自定义反序列化器；默认情况下，当读取超时(所以超时之后)时，我们关闭套接字。您必须捕获超时并返回部分消息，并提供一些信息来告知下游流返回nack。

反序列化器无法访问连接，因此无法发送nack本身。

不过，您可以在自定义子类TcpMessageMapper中这样做-重写toMessage()。

尽管如此，除非您关闭套接字，否则您的解决方案可能很脆弱，因为流可能仍然包含来自上一条消息的一些数据，尽管对于单用true，我假设每个套接字只发送一条消息。

编辑

@SpringBootApplication
public class So40408085Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So40408085Application.class, args);
        context.getBean("toTcp", MessageChannel.class).send(new GenericMessage<>("foo"));
        Thread.sleep(5000);
        context.close();
    }

    @Bean
    public TcpNetServerConnectionFactory server() {
        TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(1234);
        server.setSoTimeout(1000);
        server.setMapper(new TimeoutMapper()); // use 'mapper' attribute in XML
        return server;
    }

    @Bean
    public TcpInboundGateway inGate() {
        TcpInboundGateway inGate = new TcpInboundGateway();
        inGate.setConnectionFactory(server());
        inGate.setRequestChannelName("inChannel");
        return inGate;
    }

    @ServiceActivator(inputChannel = "inChannel")
    public String upCase(byte[] in) {
        return new String(in).toUpperCase();
    }

    @Bean
    public TcpNetClientConnectionFactory client() {
        TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", 1234);
        client.setSerializer(new ByteArrayLfSerializer()); // so the server will timeout - he's expecting CRLF
        return client;
    }

    @Bean
    @ServiceActivator(inputChannel = "toTcp")
    public TcpOutboundGateway out() {
        TcpOutboundGateway outGate = new TcpOutboundGateway();
        outGate.setConnectionFactory(client());
        outGate.setOutputChannelName("reply");
        return outGate;
    }

    @ServiceActivator(inputChannel = "reply")
    public void reply(byte[] in) {
        System.out.println(new String(in));
    }

    public static class TimeoutMapper extends TcpMessageMapper {

        @Override
        public Message<?> toMessage(TcpConnection connection) throws Exception {
            try {
                return super.toMessage(connection);
            }
            catch (SocketTimeoutException e) {
                connection.send(new GenericMessage<>("You took too long to send me data, sorry"));
                connection.close();
                return null;
            }
        }

    }

}