Spring Integration 3.0版：根据内容将TCP流消息拆分为多个Message答案

【问题标题】：Spring Integration version 3.0: Splitting TCP stream message into multiple Messages based on contentSpring Integration 3.0版：根据内容将TCP流消息拆分为多个Message
【发布时间】：2015-08-23 13:59:32
【问题描述】：

我正在使用 spring 集成框架连接到一些旧的服务器套接字。

以下是我的客户工厂和适配器：

<int-ip:tcp-connection-factory id="client"
                               type="client" 
                               host="${tcpServer}" 
                               port="${tcpPort}" 
                               single-use="false"
                               using-nio="false" />

<int-ip:tcp-inbound-channel-adapter id="inboundServer"
                                    client-mode="true"
                                    channel="inputStream" 
                                    error-channel="errorChannel"
                                    retry-interval="${retryInterval}"
                                    connection-factory="client" />

在流转字符串转换器下方：

    <int:transformer id="clientBytes2String" 
                 input-channel="inputStream"
                 output-channel="inputString" 
                 expression="new String(payload)" />

                 <int:channel id="inputString" />

下面的部分是空的，因为我不确定要在这里实现什么，以便它可以调用我的路由器，而路由器将完成它的业务。

我已经尝试使用拆分器，它确实有效，如果流以“ABCD EFGH WXYZ”或“ABCD”的所需格式出现，但如果流像“ABCD XXXX EFGH WXYZ”那样出现，那么它会失败。期望的结果是它应该处理 3 条消息和 1 个错误。但是它处理了 1 条消息，其余的都被忽略了。

以下代码：

 <int:splitter input-channel="inputString"
              output-channel="preRouter2"                  
              method="splitMessage"
              ref="messageSplitterBean"/>

和MessageSpliterBean类如下：

@Splitter
public List<Message<?>> splitMessage( Message<?> message ) {

    List<Message<?>> msgFragments = new ArrayList<Message<?>>();

    //Let say I am assuming message will be coming ABCD EFGH WXYZ
    String str[] = message.getPayload().toString().split(" ");
    int counter = 1;
    for ( String s : str ) {

        Message<String> resultMessage = MessageBuilder.withPayload( s )
                .copyHeaders(message.getHeaders())
                .build();
        msgFragments.add(resultMessage);
    }

    return msgFragments;
}

以下将是我的路由器，它将根据某些表达式发送到相应的通道：

<int:recipient-list-router id="customRouter" input-channel="preRouter2">
<int:recipient channel="input1" selector-expression="payload.toString().startsWith('ABCD')"/>   
<int:recipient channel="input2" selector-expression="payload.toString().startsWith('EFGH')"/>
<int:recipient channel="input3" selector-expression="payload.toString().startsWith('WXYZ')"/>

需要您对此的专家意见：关于我做错了什么，或者什么是最好的方法。

来自服务器套接字的输入将以固定长度和空间作为分隔符的数据流。并且每个固定长度我都需要将它们转换为消息并将其发送到相关频道。

问候。

【问题讨论】：

标签： java spring tcp spring-integration splitter

【解决方案1】：

首先没有理由实现你自己的Splitter，因为默认的有delimiters选项：

我的另一个观点是关于 byte[] -> String 的冗余 <int:transformer>。 Spring Integration 为您提供了开箱即用的ObjectToStringTransformer。

您的问题是<int:recipient-list-router>。正如您所说，您的数据可能有问题，而您的 router 还没有准备好处理这样的消息，它对您来说失败了。那只是因为（AbstractMessageRouter）：

else {
    throw new MessageDeliveryException(message, "No channel resolved by router '" + this.getComponentName()
                + "' and no 'defaultOutputChannel' defined.");
}

当没有人selector-expression 接受您的错误消息时会发生这种情况。

在这种情况下，它会发送到您的<int-ip:tcp-inbound-channel-adapter> 上的error-channel="errorChannel"，并且它会停止进一步的处理，因为MessageDeliveryException。

如您所见，您应该将default-output-channel 添加到recipient-list-router 配置中。

【讨论】：