java - 在 single-use=false 中关闭套接字连接 Spring Integration TCP Server

Question

我知道 spring 集成有 TcpInboundGateway 和 ByteArrayStxEtxSerializer 来处理通过 TCP 端口传入的数据。

如果 TCP 服务器需要读取客户端发送的所有数据然后对其进行处理，ByteArrayStxEtxSerializer 会非常有效。 (请求和响应模型)我使用 single-use=false 以便可以在同一连接中处理多个请求。

例如，如果客户端发送 0x02AAPL0x03，则服务器可以发送 AAPL 价格。

如果客户端发送 0x02AAPL0x030x02GOOG0x03，则我的 TCP 服务器正在工作。它发送 AAPL 价格和 GOOG 价格。

有时客户端可以发送 EOT (0x04)。如果客户端发送EOT，我想关闭套接字连接。

例如:客户端请求可以是0x02AAPL0x030x02GOOG0x03 0x020x040x03。注意 EOT 出现在最后一个数据包中。

我知道ByteArrayStxEtxSerializer反序列化器可以定制来读取客户端发送的字节。

解串器是关闭套接字连接的好地方吗？如果没有，如何通知spring集成框架关闭socket连接？

请帮忙。

这是我的 Spring 配置:

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

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

    <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"/>

更新:添加 throw new SoftEndOfStreamException("Stream returned") 来关闭序列化器中的流是可行的，我可以在 EventListener 中看到 CLOSED 日志条目。当服务器关闭连接时，我期望在客户端中接收 java.io.InputStream.read() 作为 -1 。但客户端正在接收

java.net.SocketTimeoutException: Read timed out
    at java.net.SocketInputStream.socketRead0(Native Method)
    at java.net.SocketInputStream.read(SocketInputStream.java:129)
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:264)
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:306)
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:158)
    at sun.nio.cs.StreamDecoder.read0(StreamDecoder.java:107)
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:93)
    at java.io.InputStreamReader.read(InputStreamReader.java:151)

还有什么可以关闭服务器端的连接并将其传播到客户端吗？

感谢您的帮助。

谢谢

Answer 1

解串器无法访问套接字，只能访问输入流；关闭它可能会起作用，但您可能会在日志中收到很多噪音。

最好的解决方案是抛出 SoftEndOfStreamException ；这表明套接字应该关闭并清理所有内容。

编辑

添加监听器来检测/记录关闭...

@SpringBootApplication
public class So40471456Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So40471456Application.class, args);
        Socket socket = SocketFactory.getDefault().createSocket("localhost", 1234);
        socket.getOutputStream().write("foo\r\n".getBytes());
        socket.close();
        Thread.sleep(10000);
        context.close();
    }

    @Bean
    public EventListener eventListener() {
        return new EventListener();
    }

    @Bean
    public TcpNetServerConnectionFactory server() {
        return new TcpNetServerConnectionFactory(1234);
    }

    @Bean
    public TcpReceivingChannelAdapter inbound() {
        TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
        adapter.setConnectionFactory(server());
        adapter.setOutputChannelName("foo");
        return adapter;
    }

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

    public static class EventListener implements ApplicationListener<TcpConnectionCloseEvent> {

        private final Log logger = LogFactory.getLog(getClass());

        @Override
        public void onApplicationEvent(TcpConnectionCloseEvent event) {
            logger.info(event);
        }

    }

}

使用 XML，只需为您的监听器类添加 <bean/>。

结果:

foo
2016-11-07 16:52:04.133  INFO 29536 --- [pool-1-thread-2] c.e.So40471456Application$EventListener  : TcpConnectionCloseEvent 
[source=org.springframework.integration.ip.tcp.connection.TcpNetConnection@118a7548], 
[factory=server, connectionId=localhost:50347:1234:b9fcfaa9-e92c-487f-be59-1ed7ebd9312e] 
**CLOSED**

编辑2

它按我的预期工作......

@SpringBootApplication
public class So40471456Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So40471456Application.class, args);
        Socket socket = SocketFactory.getDefault().createSocket("localhost", 1234);
        socket.getOutputStream().write("foo\r\n".getBytes());
        try {
            System.out.println("\n\n\n" + socket.getInputStream().read() + "\n\n\n");
            context.getBean(EventListener.class).latch.await(10, TimeUnit.SECONDS);
        }
        finally {
            socket.close();
            context.close();
        }
    }

    @Bean
    public EventListener eventListener() {
        return new EventListener();
    }

    @Bean
    public TcpNetServerConnectionFactory server() {
        TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(1234);
        server.setDeserializer(is -> {
            throw new SoftEndOfStreamException();
        });
        return server;
    }

    @Bean
    public TcpReceivingChannelAdapter inbound() {
        TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
        adapter.setConnectionFactory(server());
        adapter.setOutputChannelName("foo");
        return adapter;
    }

    public static class EventListener implements ApplicationListener<TcpConnectionCloseEvent> {

        private final Log logger = LogFactory.getLog(getClass());

        private final CountDownLatch latch = new CountDownLatch(1);

        @Override
        public void onApplicationEvent(TcpConnectionCloseEvent event) {
            logger.info(event);
            latch.countDown();
        }

    }

}

结果:

2016-11-08 08:27:25.964  INFO 86147 --- [           main] com.example2.So40471456Application       : Started So40471456Application in 1.195 seconds (JVM running for 1.764)



-1



2016-11-08 08:27:25.972  INFO 86147 --- [pool-1-thread-2] c.e.So40471456Application$EventListener  : TcpConnectionCloseEvent [source=org.springframework.integration.ip.tcp.connection.TcpNetConnection@fee3774], [factory=server, connectionId=localhost:54984:1234:f79a6826-0336-4823-8844-67054903a094] **CLOSED**