如何在spring集成TCP Server中等待数据

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

你需要一个自定义的反序列化器；默认情况下，当读取超时（在 so-timeout 之后）我们关闭套接字。您必须捕获超时并返回部分消息，其中包含一些信息来告诉下游流返回 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;
            }
        }

    }

}