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等)之一。