如何在 Spring XD 中的 "side channel" 上发送消息？

Question

我想实现这样的目标：

[Source] --> [Splitter] --B--> [Processor] --B-->  [Aggregator] ---> [Sink]
              |                                        ^
              |                                        |
              |--------------------C-------------------|

经过几个包含业务逻辑的处理器之后，一个特殊的生产者应该发出几个类型 B 的消息以及一个控制消息 C.

具有更多业务逻辑的处理器将处理这些消息。

聚合器应"collect"所有结果以及控制消息。

控制消息基本上包含生成消息（B）的数量，因此聚合器可以知道是否

所有预期的 B 消息都已到达
发生超时

我知道，聚合器可以通过使用消息头推断出拆分器期望的消息数量，但是如果没有消息到达它，它也会超时吗？

一般来说：Transformer 是否可以有一个以上的输入或输出通道？

Answer 1

2先-

标准聚合器模块有一个 timeout 参数（默认为 50 秒）。

它是通过每超时秒运行一次的消息组存储收割器实现的 - 因此实际超时将达到 2 x timeout，平均为 1.5 x timeout。

聚合器现在有一个 group-timeout 属性，它比使用 reaear 更准确；它需要一个自定义聚合器处理器才能使用它。它还有一个 group-timeout-expression，因此超时可以根据运行时条件（例如当前组大小）而变化。

现在 1 -

没有标准机制可以将 side-band 数据从一个模块发送到另一个模块。一般模块之间通过设置messageheaders进行通信。这就是标准拆分器向聚合器发送信息的方式（sequenceNumber 和 sequenceSize headers）。

您可以使用自定义 ReleaseStrategy 创建自定义聚合器处理器以使用其他 headers。

Generally speaking: is it possible to have Transformers with more than one input or output channel?

不支持 XD，但 follow-on 项目 (spring-cloud-stream) 支持将多个 inputs/outputs 绑定到应用程序。

如何在 Spring XD 中的 "side channel" 上发送消息？

How to send messages on a "side channel" in Spring XD?

java

spring

spring-integration

spring-xd