Spring Integration Java DSL TCP客户端无法接收服务器响应排查求助

问题现状 #

已实现动态TCP客户端Outbound Adapter，请求可正常发送至服务器，且通过Wireshark确认服务器已返回响应，但代码无法捕获该响应，需排查代码问题。

代码分析与排查点 #

1. 连接工厂与集成流代码问题 #

TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory(ip, port);
cf.setDeserializer(new CustomDeserializer());

IntegrationFlow out = IntegrationFlow.from("clientInputChannel")
        .handle(Tcp.outboundAdapter(cf))
        .transform(Transformers.objectToString())
        .get();

this.flowContext.registration(out)
        .addBean(cf)
        .id(port + ".out")
        .register();

IntegrationFlow in = IntegrationFlow.from(Tcp.inboundAdapter(cf)).channel("fromTcp").get();

this.flowContext.registration(in)
        .id(port + ".in")
        .register();

IntegrationFlow responseFlow = IntegrationFlow.from("fromTcp")
        .handle(message -> {
            log.info("UGV Response " + message.getPayload());
        })
        .get();

this.flowContext.registration(responseFlow)
        .id("errorFlow")
        .register();

• 连接工厂复用错误：同一个TcpNetClientConnectionFactory同时绑定给Outbound Adapter和Inbound Adapter是错误的。Inbound Adapter用于服务器端监听端口接收连接，客户端场景下无需单独配置Inbound Adapter，响应应由Outbound Adapter对应的连接工厂自动处理。
• 冗余Transform步骤：Tcp.outboundAdapter是单向发送组件，无返回值，后续的.transform(Transformers.objectToString())不会被执行，属于无效代码，应删除。
• 响应路由逻辑错误：单向Outbound Adapter不会自动将响应路由到指定通道，若需处理响应，应改用请求-响应模式的Tcp.outboundGateway。

2. 消息通道代码 #

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

@Bean("fromTcp")
public QueueChannel fromTcp() {
    return new QueueChannel();
}

• QueueChannel本身无问题，但需确认是否有消费者及时处理消息。可添加通道拦截器日志，验证是否有消息进入该通道。

3. 消息发送代码 #

@Autowired
@Qualifier("clientInputChannel")
private MessageChannel toTcp;

@Override
public boolean sendHex(byte[] buffer) {
   log.info("Sending Bytes: " + buffer);
    toTcp.send(MessageBuilder.withPayload(buffer).build());
   log.info("Bytes Sent");
   return true;
}

• 发送逻辑无明显问题，字节数组可正常发送至服务器。

4. 测试用ServiceActivator代码 #

@ServiceActivator(inputChannel = "fromTcp")
public void receiver(PelcoResponseMessage in) {
    System.out.println("Received " + in + " from ");
}

@ServiceActivator(inputChannel = "fromTcp")
public void receiver1(byte[] in) {
    System.out.println("Received1 " + in + " from ");
}

@ServiceActivator(inputChannel = "fromTcp")
public void receiver2(String in) {
    System.out.println("Received2 " + in + " from ");
}

• 类型匹配风险：若CustomDeserializer返回的对象类型与所有ServiceActivator的参数类型不匹配，消息会无法被处理。需确认反序列化后的payload类型，是否与某一个接收器的参数类型一致。

解决方案建议 #

1. 替换为Tcp.outboundGateway处理请求-响应 #

将单向的Tcp.outboundAdapter改为请求-响应模式的Tcp.outboundGateway，自动将响应路由到指定通道，无需单独配置Inbound Adapter：

IntegrationFlow out = IntegrationFlow.from("clientInputChannel")
        .handle(Tcp.outboundGateway(cf))
        .channel("fromTcp")
        .get();

this.flowContext.registration(out)
        .addBean(cf)
        .id(port + ".out")
        .register();

// 移除原有的Inbound Adapter配置

2. 验证CustomDeserializer实现 #

确保反序列化器能正确解析服务器返回的字节流，添加日志确认解析过程：

public class CustomDeserializer implements Deserializer<Object> {
    private static final Logger log = LoggerFactory.getLogger(CustomDeserializer.class);

    @Override
    public Object deserialize(InputStream inputStream) throws IOException {
        byte[] buffer = new byte[1024];
        int readBytes = inputStream.read(buffer);
        if (readBytes > 0) {
            byte[] responseBytes = Arrays.copyOf(buffer, readBytes);
            log.info("Received response bytes: {}", Arrays.toString(responseBytes));
            // 根据协议转换为PelcoResponseMessage，或直接返回byte[]
            return convertToPelcoResponseMessage(responseBytes);
        }
        log.warn("No bytes read from input stream");
        return null;
    }
}

3. 添加通道日志验证消息流入 #

在fromTcp通道添加拦截器，确认是否有消息进入：

@Bean("fromTcp")
public QueueChannel fromTcp() {
    QueueChannel channel = new QueueChannel();
    channel.addInterceptor(new ChannelInterceptor() {
        @Override
        public Message<?> preSend(Message<?> message, MessageChannel channel) {
            log.info("Message entered fromTcp channel: payload type={}, content={}",
                    message.getPayload().getClass().getName(), message.getPayload());
            return message;
        }
    });
    return channel;
}

4. 清理冗余代码 #

删除Outbound Adapter后的无效transform步骤，精简集成流配置。

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