Spring集成TCP服务器未接收消息

Question

我正在尝试创建一个TCP服务器，该服务器在端口5002上接受来自外部程序的消息。但是，它不接收来自外部程序的消息。

@Bean
public TcpReceivingChannelAdapter inbound(AbstractServerConnectionFactory cf) {
   TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
   adapter.setConnectionFactory(cf);
   adapter.setOutputChannel(tcpIn());
   return adapter;
 }

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

@Bean
@Transformer(inputChannel = "tcpIn", outputChannel = "serviceChannel")
public ObjectToStringTransformer transformer() {
    return new ObjectToStringTransformer();
}

@ServiceActivator(inputChannel = "serviceChannel")
public void messageToService(String in) {
    // Message received
}

@Bean
public AbstractServerConnectionFactory serverConnectionFactory() {
    TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(5002);
    tcpNetServerConnectionFactory.setSoTimeout(5000);
    tcpNetServerConnectionFactory.setMapper(new TimeoutMapper());
    return tcpNetServerConnectionFactory;
}

为了验证我的TCP服务器是否正常工作，我像这样使用了telnet，程序确实收到了文本“hello”。

telnet 192.168.1.2 5002
Trying 192.168.1.2...
Connected to 192.168.1.2.
Escape character is '^]'.
hello

设置wireshark时，我可以看到计算机正在端口5002上接收来自外部程序（我期待）的消息。为什么我的程序无法接收这些消息？

关于最终解决方案的最新情况：

由于负载没有停止线，我必须实现我自己的反序列化器，如@ArtemBilan所述。我使用~字符来表示客户端的行结束。

@Bean
public AbstractServerConnectionFactory serverConnectionFactory() {
    TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(tcpPort);
    tcpNetServerConnectionFactory.setSoTimeout(0);
  tcpNetServerConnectionFactory.setDeserializer(endOfLineSerializer());
    tcpNetServerConnectionFactory.setSerializer(endOfLineSerializer());
    tcpNetServerConnectionFactory.setMapper(new TimeoutMapper());
    return tcpNetServerConnectionFactory;
}

我实现的示例序列化器：

public class EndOfLineSerializer extends AbstractPooledBufferByteArraySerializer {

private static final char MANUAL_STOP_LINE = '~';
private static final char AUTO_STOP_LINE = '\t';
private static final byte[] CRLF = "\r\n".getBytes();

/**
 * Reads the data in the inputStream to a byte[]. Data must be terminated
 * by a single byte. Throws a {@link SoftEndOfStreamException} if the stream
 * is closed immediately after the terminator (i.e. no data is in the process of
 * being read).
 */
@Override
protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
    int n = 0;
    int bite;

    try {
        while (true) {

            try {
                bite = inputStream.read();
            } catch (SocketTimeoutException e) {
                bite = -1;
            }

            if (bite < 0) {
                // Payload complete
                break;
            }

            if ((n > 0 && bite == '\n' && buffer[n - 1] == '\r') || bite == this.MANUAL_STOP_LINE || bite == this.AUTO_STOP_LINE) {
                break;
            }

            buffer[n++] = (byte) bite;
            if (n >= this.maxMessageSize) {
                throw new IOException("Terminator not found before max message length: " + this.maxMessageSize);
            }
        }
        return copyToSizedArray(buffer, n);
    } catch (IOException e) {
        publishEvent(e, buffer, n);
        throw e;
    } catch (RuntimeException e) {
        publishEvent(e, buffer, n);
        throw e;
    }
}

/**
 * Writes the byte[] to the stream and appends the CRLF.
 */
@Override
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
    outputStream.write(bytes);
    outputStream.write(this.CRLF);
    }
}

Answer 1

见文件；向下滚动至

TCP是一种流协议；这意味着必须为通过TCP传输的数据提供某种结构，以便接收方可以将数据划分为离散消息。连接工厂配置为使用（反）序列化器在消息有效负载和通过TCP发送的位之间进行转换。这是通过分别为入站和出站消息提供反序列化器和序列化器来实现的。提供了许多标准（反）序列化程序。

并阅读有关标准反序列化程序的信息。根据您的配置，标准反序列化程序正在等待终止符\r\n（CRLF）。

Telnet附加了CRLF，这就是它工作的原因。

Answer 2

默认情况下，TcpNetServerConnectionFactory使用ByteArrayCrLfSerializer，其中这是作为消息分隔符计数的内容：

private static final byte[] CRLF = "\r\n".getBytes();

因此，您应该确保您的客户机在最后发送带有正确符号的消息。

有许多现成的序列化程序供您选择：

https://docs.spring.io/spring-integration/docs/5.0.3.RELEASE/reference/html/ip.html#tcp-连接工厂

或者，您可以实现自己的Deserializer并将其注入到serverConnectionFactorybean定义中。