Spring Cloud Stream支持将多个应用程序聚合在一起，直接连接其输入和输出通道，并避免通过代理交换消息的额外成本。从版本1.0的Spring Cloud Stream开始，仅对以下类型的应用程序支持聚合：
来源- 具有名为output的单个输出通道的应用程序，通常具有类型为org.springframework.cloud.stream.messaging.Source的单个绑定
接收器- 具有名为input的单个输入通道的应用程序，通常具有类型为org.springframework.cloud.stream.messaging.Sink的单个绑定
处理器- 具有名为input的单个输入通道和名为output的单个输出通道的应用程序，通常具有类型为org.springframework.cloud.stream.messaging.Processor的单个绑定。
它们可以通过创建一系列互连的应用程序来聚合在一起，其中序列中的元素的输出通道连接到下一个元素的输入通道（如果存在）。序列可以从源或处理器开始，它可以包含任意数量的处理器，并且必须以处理器或接收器结束。
根据起始和结束元素的性质，序列可以具有一个或多个可绑定的信道，如下所示：
如果序列从源头开始并以sink结束，则应用程序之间的所有通信都是直接的，并且不会绑定任何通道
如果序列以处理器开始，则其输入通道将成为聚合的input通道，并将相应地进行绑定
如果序列以处理器结束，则其输出通道将成为聚合的output通道，并将相应地进行绑定
使用AggregateApplicationBuilder实用程序类执行聚合，如以下示例所示。我们考虑一个项目，我们有源，处理器和汇点，可以在项目中定义，或者可以包含在项目的依赖之一中。
注意如果配置类使用@SpringBootApplication，聚合应用程序中的每个组件（源，宿或处理器）必须在单独的包中提供。这是为了避免应用程序之间的串扰，由于在同一个包内的配置类上由@SpringBootApplication执行的类路径扫描。在下面的示例中，可以看到，Source，Processor和Sink应用程序类分组在单独的包中。一个可能的替代方法是在单独的@Configuration类中提供源，宿或处理器配置，避免使用@SpringBootApplication/@ComponentScan并使用它们进行聚合。
package com.app.mysink;
@SpringBootApplication
@EnableBinding(Sink.class)
public class SinkApplication {
private static Logger logger = LoggerFactory.getLogger(SinkApplication.class);
@ServiceActivator(inputChannel=Sink.INPUT)
public void loggerSink(Object payload) {
logger.info("Received: " + payload);
}
}
package com.app.myprocessor;
@SpringBootApplication
@EnableBinding(Processor.class)
public class ProcessorApplication {
@Transformer
public String loggerSink(String payload) {
return payload.toUpperCase();
}
}
package com.app.mysource;
@SpringBootApplication
@EnableBinding(Source.class)
public class SourceApplication {
@Bean
@InboundChannelAdapter(value = Source.OUTPUT)
public String timerMessageSource() {
return new SimpleDateFormat().format(new Date());
}
}
每个配置可以用于运行一个单独的组件，但在这种情况下，它们可以聚合在一起，如下所示：
package com.app;
@SpringBootApplication
public class SampleAggregateApplication {
public static void main(String[] args) {
new AggregateApplicationBuilder()
.from(SourceApplication.class).args("--fixedDelay=5000")
.via(ProcessorApplication.class)
.to(SinkApplication.class).args("--debug=true").run(args);
}
}
该序列的起始组件作为from()方法的参数提供。序列的结尾部分作为to()方法的参数提供。中间处理器作为via()方法的参数提供。同一类型的多个处理器可以一起链接（例如，用于具有不同配置的流水线转换）。对于每个组件，构建器可以为Spring Boot配置提供运行时参数。
配置聚合应用程序
Spring Cloud Stream支持使用'namespace'作为前缀向聚合应用程序内的各个应用程序传递属性。
命名空间可以为应用程序设置如下：
@SpringBootApplication
public class SampleAggregateApplication {
public static void main(String[] args) {
new AggregateApplicationBuilder()
.from(SourceApplication.class).namespace("source").args("--fixedDelay=5000")
.via(ProcessorApplication.class).namespace("processor1")
.to(SinkApplication.class).namespace("sink").args("--debug=true").run(args);
}
}
一旦为单个应用程序设置了“命名空间”，则可以使用任何支持的属性源（命令行，环境属性等）将具有namespace作为前缀的应用程序属性传递到聚合应用程序，
例如，要覆盖“source”和“sink”应用程序的默认fixedDelay和debug属性：
java -jar target/MyAggregateApplication-0.0.1-SNAPSHOT.jar --source.fixedDelay=10000 --sink.debug=false
为非自包含聚合应用程序配置绑定服务属性
非自包含聚合应用程序通过聚合应用程序的入站/出站组件（通常为消息通道）中的一个或两者绑定到外部代理，而聚合应用程序内的应用程序是直接绑定的。例如：源应用程序的输出和处理器应用程序的输入是直接绑定的，而处理器的输出通道绑定到代理的外部目的地。当传递非自包含聚合应用程序的绑定服务属性时，需要将绑定服务属性传递给聚合应用程序，而不是将它们设置为单个子应用程序的“args”。例如，
@SpringBootApplication
public class SampleAggregateApplication {
public static void main(String[] args) {
new AggregateApplicationBuilder()
.from(SourceApplication.class).namespace("source").args("--fixedDelay=5000")
.via(ProcessorApplication.class).namespace("processor1").args("--debug=true").run(args);
}
}
需要将绑定属性--spring.cloud.stream.bindings.output.destination=processor-output指定为外部配置属性（cmdline arg等）之一。
