如何使用 TcpInboundGateway 响应请求？答案

【问题标题】：How to respond to request with TcpInboundGateway?如何使用 TcpInboundGateway 响应请求？
【发布时间】：2017-10-02 12:48:30
【问题描述】：

我想用 Spring Integration 开发一个 TCP 服务器，但我有一些困难...... 我的服务器应该侦听 GPS 设备的端口，这些设备首先通过 IMEI 识别自己，需要来自服务器的确认才能进一步将位置数据发送到同一个 TCP 服务器。

我成功获取了 IMEI，但确认没有到达设备...我什至不确定我是否以正确的方式发送它。我阅读了文档和其他帖子，发现我必须使用 TcpInboundGateway 而不是通道适配器，因为它是同一个会话。但是我必须使用TcpOutboundGateway 将响应发送回设备吗？

这是我的代码：

@EnableIntegration
@IntegrationComponentScan
@Configuration
public class TcpServerConfiguration {

    private static final Logger logger = LogManager.getLogger(TcpServerConfiguration.class);

    @Autowired
    TcpServerProperties properties;

    @Bean
    public AbstractServerConnectionFactory serverFactory() {
        TcpNetServerConnectionFactory tcpServerFactory = new TcpNetServerConnectionFactory(properties.getPort());
        tcpServerFactory.setDeserializer(new ByteArrayRawSerializer());
        return tcpServerFactory;
    }

    @Bean
    public MessageChannel inputChannel() {
        DirectChannel channel = new DirectChannel();
        return channel;
    }

    @Bean
    public MessageChannel outputChannel() {
        DirectChannel channel = new DirectChannel();
        return channel;
    }

    @Bean
    public TcpInboundGateway tcpInGate(AbstractServerConnectionFactory connectionFactory)  {
        TcpInboundGateway inGateway = new TcpInboundGateway();
        inGateway.setConnectionFactory(connectionFactory);
        inGateway.setRequestChannel(inputChannel());
        inGateway.setReplyChannel(outputChannel());
        return inGateway;
    }

}

消息处理类：

@MessageEndpoint
public class MessageHandler {

    private static final Logger logger = LogManager.getLogger(MessageHandler.class);

    @Transformer( inputChannel = "inputChannel", outputChannel = "outputChannel")
    public byte[] consume(byte[] bytes) {
        String message = new String(bytes);
        String byteString = "";
        for (byte b : bytes) {
            byteString += Byte.toString(b) + ", ";
        }
        logger.info("Bytes : " + byteString);
        logger.info("Message : " + message);
        return new byte[]{ 01 };
    }

}

这里我总是发送 Ack (01) 来测试它是否发送成功，但事实并非如此。

我添加了logging.level.org.springframework.integration=DEBUG，这是我得到的：

2017-10-02 14:16:44.245  INFO 6340 --- [main] b.thingsplay.Fmb920TcpServerApplication  : Started Fmb920TcpServerApplication in 3.885 seconds (JVM running for 5.38)
2017-10-02 14:16:44.250  INFO 6340 --- [pool-1-thread-1] .s.i.i.t.c.TcpNetServerConnectionFactory : serverFactory, port=7015 Listening
2017-10-02 14:17:50.161 DEBUG 6340 --- [pool-1-thread-1] .s.i.i.t.c.TcpNetServerConnectionFactory : Accepted connection from 1.1.1.1
2017-10-02 14:17:50.200 DEBUG 6340 --- [pool-1-thread-1] o.s.i.i.tcp.connection.TcpNetConnection  : New connection ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912
2017-10-02 14:17:50.201 DEBUG 6340 --- [pool-1-thread-1] .s.i.i.t.c.TcpNetServerConnectionFactory : serverFactory: Added new connection: ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912
2017-10-02 14:17:50.204 DEBUG 6340 --- [pool-1-thread-2] o.s.i.i.tcp.connection.TcpNetConnection  : ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912 Reading...
2017-10-02 14:17:50.207 DEBUG 6340 --- [pool-1-thread-2] o.s.i.i.t.s.ByteArrayRawSerializer       : Available to read:17
2017-10-02 14:18:20.576 DEBUG 6340 --- [pool-1-thread-2] o.s.i.i.tcp.connection.TcpNetConnection  : Message received GenericMessage [payload=byte[17], headers={ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=229435ad-6284-8136-40d4-4fdfe3ef462e, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700575}]
2017-10-02 14:18:20.585  INFO 6340 --- [pool-1-thread-2] o.s.i.endpoint.EventDrivenConsumer       : Adding {bridge:null} as a subscriber to the 'outputChannel' channel
2017-10-02 14:18:20.587  INFO 6340 --- [pool-1-thread-2] o.s.integration.channel.DirectChannel    : Channel 'application.outputChannel' has 1 subscriber(s).
2017-10-02 14:18:20.589  INFO 6340 --- [pool-1-thread-2] o.s.i.endpoint.EventDrivenConsumer       : started org.springframework.integration.endpoint.EventDrivenConsumer@14c3d44a
2017-10-02 14:18:20.595 DEBUG 6340 --- [pool-1-thread-2] o.s.integration.channel.DirectChannel    : preSend on channel 'inputChannel', message: GenericMessage [payload=byte[17], headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=2cc3d793-dc24-705d-1438-d4b3e089d11b, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700594}]
2017-10-02 14:18:20.598 DEBUG 6340 --- [pool-1-thread-2] o.s.i.t.MessageTransformingHandler       : messageHandler.consume.transformer.handler received message: GenericMessage [payload=byte[17], headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=2cc3d793-dc24-705d-1438-d4b3e089d11b, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700594}]
2017-10-02 14:18:20.608  INFO 6340 --- [pool-1-thread-2] be.thingsplay.tcp.MessageHandler         : Bytes : 0, 15, 51, 53, 50, 48, 57, 52, 48, 56, 51, 50, 54, 54, 52, 55, 53, 
2017-10-02 14:18:20.611  INFO 6340 --- [pool-1-thread-2] be.thingsplay.tcp.MessageHandler         : Message :  352094083266475
2017-10-02 14:18:20.618 DEBUG 6340 --- [pool-1-thread-2] o.s.integration.channel.DirectChannel    : preSend on channel 'outputChannel', message: GenericMessage [payload=byte[1], headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=0e5b6c37-ff73-5c46-4c37-0df48d3ba607, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700614}]
2017-10-02 14:18:20.620 DEBUG 6340 --- [pool-1-thread-2] o.s.integration.handler.BridgeHandler    : org.springframework.integration.handler.BridgeHandler@5b393299 received message: GenericMessage [payload=byte[1], headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=0e5b6c37-ff73-5c46-4c37-0df48d3ba607, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700614}]
2017-10-02 14:18:20.623 DEBUG 6340 --- [pool-1-thread-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'outputChannel', message: GenericMessage [payload=byte[1], headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=0e5b6c37-ff73-5c46-4c37-0df48d3ba607, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700614}]
2017-10-02 14:18:20.625 DEBUG 6340 --- [pool-1-thread-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel 'inputChannel', message: GenericMessage [payload=byte[17], headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@1a0438a2, ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=2cc3d793-dc24-705d-1438-d4b3e089d11b, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700594}]
2017-10-02 14:18:20.628 DEBUG 6340 --- [pool-1-thread-2] o.s.i.i.tcp.connection.TcpNetConnection  : ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912 Message sent GenericMessage [payload=byte[1], headers={ip_tcp_remotePort=62138, ip_connectionId=ptr-x-x-x-x.dyn.mobistar.be:62138:7015:a7151dbf-63c1-4ff5-b158-5707797b8912, ip_localInetAddress=/0.0.0.0, ip_address=1.1.1.1, id=af87b495-fceb-dbdc-c768-a8a81391af75, ip_hostname=ptr-x-x-x-x.dyn.mobistar.be, timestamp=1506946700628}]
2017-10-02 14:18:20.631 DEBUG 6340 --- [pool-1-thread-2] o.s.i.i.t.s.ByteArrayRawSerializer       : Available to read:0

希望有人能帮助我！

【问题讨论】：

标签： java spring tcp spring-integration

【解决方案1】：

你使用TcpInboundGateway的正确方式。尽管您可能不需要setReplyChannel() 选项。在大多数情况下，完全依赖标头中的 replyChannel 就足够了，并且从 @Transformer 方法返回的 jsut 将到达正确的“会话”进行回复。

我认为你使用了错误的deserializer。见ByteArrayRawSerializerJavaDocs：

 * A byte array (de)serializer that does nothing with the payload; sends it raw.
 * Message termination for assembly purposes is signaled by the client closing the
 * connection. The serializer does not, itself, close the connection after
 * writing the bytes.
 * <p>
 * Because the socket must be closed to indicate message end, this (de)serializer
 * can only be used by uni-directional (non-collaborating) channel adapters, and
 * not by gateways.

因此，由于您的客户端没有关闭结束会话的连接，因此您不会在那里得到回复。此序列化程序不适用于网关。

考虑为您选择合适的（反）序列化器和适当的终结器：https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/ip.html#connection-factories

【讨论】：

谢谢，确实是这个问题。但是当没有终止值时，我可以使用什么作为序列化程序？从文档中，我的请求如下所示：'例如，IMEI 356307042441013 将作为 000f333536333037303432343431303133 发送前两个字节表示 IMEI 长度。在这种情况下，000F 意味着，imei 的长度为 15 个字节。'
我不知道。您必须与您的客户协商以同意适当的消息协议。