SpringIntegration——到底是谁“迭代”了来自拆分器的拆分消息？

Question

关于SI中的拆分器元素，我有几个基本问题。

我知道要形成定制的拆分器逻辑，我们需要扩展AbstractMessageSplitter并覆盖splitMessage方法。

然后，这些拆分消息的集合将显示在拆分器的输出通道上（假设在传入消息上没有配置回复通道）。

分离器的输出通道随后可以成为另一si组件的输入通道。让我们假设该组件是一个简单的服务激活器。

手册说集合的“每个”消息将由呈现该集合的下游配置处理。所以在我们的例子中，service-tivator将处理每个消息。

1） 我的查询是“谁”确切地迭代集合，并在ServiceActivator的输入通道上显示集合中的“每条”消息？

2）假设我们这里涉及到所有的直接渠道。现在，如果service-tivator的outout-Channel被配置为nullChannel，那么一旦第一条消息被成功处理，它仍然会显示来自集合的"下一条"消息吗？

3） 我们假设这里没有聚合器。另外，我们假设所有涉及的方法都具有非void返回。将消息“收集”到拆分器的输出通道的调用线程将返回什么？它是否会返回service activator返回的返回类型集合？或者它只会从收集中返回最后处理的消息吗？

我希望我提出的问题足够详细，以便得到澄清。

非常感谢回应。

最好的问候




Answer 1

是的。你问的问题很好！

1） AbstractMessageProducingHandler：

protected void sendOutputs(Object result, Message<?> requestMessage) {
    if (result instanceof Iterable<?> && shouldSplitOutput((Iterable<?>) result)) {
        for (Object o : (Iterable<?>) result) {
            this.produceOutput(o, requestMessage);
        }
    }
    else if (result != null) {
        this.produceOutput(result, requestMessage);
    }
}

正是从这里，每个项目都被发送到输出通道。

2）是的，如果一切都在DirectChannel上，下一条消息直到完成前一条消息才会发送。实际上，它的工作方式与原始java相同：

 for (Object o : collection) {

 }

注意：在此sync情况下，如果当前消息进程失败，则不会发送下一个消息。因为异常。就像原始Java：-）。

3） 嗯。对于原始Java，它可能看起来像：

split(Collection<?>, Future<?>);  
 --->>
 for (Object o : collection) {
     process(Object, Future<?>)
 }

确切地说，这个过程是在拆分器中完成的，或者在每个EIP组件中完成得更好。实际上，它们都只在推送模式下工作。即使对回复做了一些操作，实际上也只是推送到replyChannel。

因此，如果您不关心回复过程，那么只有第一条消息将填充该将来的回复

注意，如果您的下游拆分器进程是异步的，例如，执行器通道作为输出通道打开

我认为最后一个问题的答案取决于目标业务逻辑需求。