Spring Integration中TcpInboundGateway回复失败问题求助

问题 #

使用场景 #

• 客户端通过netcat向服务器发送TCP套接字请求

  echo -n "coincoincoin" | nc -v host port
  

• 基于Spring Integration的TCP服务器接收客户端请求
• 服务器通过TcpInboundGateway接收请求并处理
• 服务器通过TcpInboundGateway向客户端返回回复
• 客户端接收并显示回复

步骤1-3可正常执行，但步骤4失败，报错：

... Failed to send reply ...

服务器配置 #

@Configuration
public class ServerConfig {

    @Value("${server.port}")
    private int port;

   @Bean
    TcpNetServerConnectionFactory serverConnexion() {
        TcpNetServerConnectionFactory factory = new TcpNetServerConnectionFactory(port);
        factory.setSingleUse(true);
        factory.setDeserializer(new CustomSerializerDeserializer());
        factory.setSerializer(new CustomSerializerDeserializer());
       factory.setSoTimeout(10000);
        return factory;
    }

    @Bean
    TcpInboundGateway tcpGateway() {
        TcpInboundGateway gateway = new TcpInboundGateway();
        gateway.setConnectionFactory(serverConnexion());
        gateway.setRequestChannel(reqChannel());
        gateway.setReplyChannel(reqChannel());
        return gateway;
    }

    @Bean
    public MessageChannel reqChannel() {
        return new DirectChannel();
    }
}

异常信息 #

2022-12-18 14:21:51.524 DEBUG 14816 --- [pool-3-thread-2] org.springframework.integration.channel.DirectChannel : [( - )] postSend (sent=true) on channel 'bean 'reqChannel'; defined in: 'class path resource [com/server/config/ServerConfig.class]'; from source: 'com.rss.server.config.ServerConfig.reqChannel()'', message: GenericMessage [payload=byte[40], headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@68980b5b, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@68980b5b, ip_tcp_remotePort=38567, ip_connectionId=kubernetes.docker.internal:38567:11000:e143aeae-532c-4aab-9ef1-ba7930ecc3ef, ip_localInetAddress=0.0.0.0/0.0.0.0, ip_address=127.0.0.1, id=3265835d-8088-7820-d7e5-0dd46ddab69c, ip_hostname=kubernetes.docker.internal, timestamp=1671369710212}]
2022-12-18 14:21:51.524 ERROR 14816 --- [pool-3-thread-2] org.springframework.integration.ip.tcp.TcpInboundGateway : [( - )] Failed to send reply
org.springframework.messaging.MessagingException: Send Failed; nested exception is java.net.SocketException: Socket is closed
    at org.springframework.integration.ip.tcp.connection.TcpNetConnection.send(TcpNetConnection.java:119) ~[spring-integration-ip-5.5.8.jar:5.5.8]
    at org.springframework.integration.ip.tcp.TcpInboundGateway.doOnMessage(TcpInboundGateway.java:139) [spring-integration-ip-5.5.8.jar:5.5.8]
    at org.springframework.integration.ip.tcp.TcpInboundGateway.onMessage(TcpInboundGateway.java:101) [spring-integration-ip-5.5.8.jar:5.5.8]
    at org.springframework.integration.ip.tcp.connection.TcpNetConnection.receiveAndProcessMessage(TcpNetConnection.java:224) [spring-integration-ip-5.5.8.jar:5.5.8]
    at org.springframework.integration.ip.tcp.connection.TcpNetConnection.run(TcpNetConnection.java:197) [spring-integration-ip-5.5.8.jar:5.5.8]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
    at java.lang.Thread.run(Thread.java:834) [?:?]
Caused by: java.net.SocketException: Socket is closed
    at java.net.Socket.getSendBufferSize(Socket.java:1215) ~[?:?]
    at org.springframework.integration.ip.tcp.connection.TcpNetConnection.send(TcpNetConnection.java:108) ~[spring-integration-ip-5.5.8.jar:5.5.8]
    ... 7 more

解决方案 #

核心问题分析 #

报错Socket is closed的根源有两点：

1. 连接工厂实例不共享：tcpGateway()方法中直接调用serverConnexion()，每次调用都会生成新的TcpNetServerConnectionFactory实例，导致网关使用的连接工厂与实际处理请求的连接工厂不一致，无法找到对应的活跃连接。
2. 单连接配置冲突：setSingleUse(true)会让连接在处理完请求后立即关闭，但网关需要复用该连接发送响应，此时连接已关闭，自然无法发送回复。

修复步骤 #

1. 注入共享连接工厂：修改tcpGateway()方法，通过依赖注入获取已创建的serverConnexion Bean，避免重复创建实例。
2. 调整连接复用配置：将factory.setSingleUse(true)改为factory.setSingleUse(false)，让连接可以复用至客户端主动关闭。

修改后的配置代码：

@Configuration
public class ServerConfig {

    @Value("${server.port}")
    private int port;

    @Bean
    TcpNetServerConnectionFactory serverConnexion() {
        TcpNetServerConnectionFactory factory = new TcpNetServerConnectionFactory(port);
        factory.setSingleUse(false); // 修改为false，允许连接复用
        factory.setDeserializer(new CustomSerializerDeserializer());
        factory.setSerializer(new CustomSerializerDeserializer());
        factory.setSoTimeout(10000);
        return factory;
    }

    @Bean
    TcpInboundGateway tcpGateway(TcpNetServerConnectionFactory serverConnexion) { // 注入已有的连接工厂Bean
        TcpInboundGateway gateway = new TcpInboundGateway();
        gateway.setConnectionFactory(serverConnexion);
        gateway.setRequestChannel(reqChannel());
        // 移除手动指定的replyChannel，默认使用请求消息中的临时回复通道即可
        return gateway;
    }

    @Bean
    public MessageChannel reqChannel() {
        return new DirectChannel();
    }
}

额外优化提示 #

• 客户端使用nc时，可添加-q 1参数（如echo -n "coincoincoin" | nc -q 1 host port），让客户端等待1秒后再关闭连接，确保能完整接收服务器回复。

内容的提问来源于stack exchange，提问作者bacalum