Spring Cloud Stream还支持使用反应性API，其中将传入和传出数据作为连续数据流处理。通过spring-cloud-stream-reactive提供对反应性API的支持，需要将其明确添加到您的项目中。
具有反应性API的编程模型是声明式的，而不是指定如何处理每个单独的消息，您可以使用描述从入站到出站数据流的功能转换的运算符。
Spring Cloud Stream支持以下反应性API：
反应堆
RxJava 1.x
将来，它旨在支持基于活动流的更通用的模型。
反应式编程模型还使用@StreamListener注释来设置反应处理程序。差异在于：
@StreamListener注释不能指定输入或输出，因为它们作为参数提供，并从方法返回值;
必须使用@Input和@Output注释方法的参数，指示输入和分别输出的数据流连接到哪个输入或输出;
方法的返回值（如果有的话）将用@Output注释，表示要发送数据的输入。
注意反应式编程支持需要Java 1.8。
注意截至Spring Cloud Stream 1.1.1及更高版本（从布鲁克林发行版开始列出），反应式编程支持需要使用Reactor 3.0.4.RELEASE和更高版本。不支持早期的Reactor版本（包括3.0.1.RELEASE，3.0.2.RELEASE和3.0.3.RELEASE）。spring-cloud-stream-reactive将会过渡地检索正确的版本，但项目结构有可能将io.projectreactor:reactor-core的版本管理到较早版本，特别是在使用Maven时。对于通过Spring Initializr（Spring Boot 1.x）生成的项目，这将覆盖Reactor版本为2.0.8.RELEASE。在这种情况下，您必须确保释放正确版本的工件。这可以通过在io.projectreactor:reactor-core上直接依赖于3.0.4.RELEASE或更高版本的项目来简单地实现。
注意术语reactive的使用目前指的是正在使用的反应性API，而不是执行模型是无效的（即，绑定的端点仍然使用“推”而不是“拉”模型）。虽然通过使用Reactor提供了一些背压支持，但我们希望长期以来通过使用连接的中间件的本机反应客户端来支持完全无反应的管道。
基于反应器的处理程序
基于反应器的处理程序可以具有以下参数类型：
对于用@Input注释的参数，它支持反应器类型Flux。入站通量的参数化遵循与单个消息处理相同的规则：它可以是整个Message，一个可以是Message有效负载的POJO，也可以是一个POJO基于Message内容类型头的转换。提供多个输入;
对于使用Output注释的参数，它支持将方法生成的Flux与输出连接的类型FluxSender。一般来说，仅当方法可以有多个输出时才建议指定输出作为参数;
基于反应器的处理程序支持Flux的返回类型，其中必须使用@Output注释。当单个输出通量可用时，我们建议使用该方法的返回值。
这是一个简单的基于反应器的处理器的例子。
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {
@StreamListener
@Output(Processor.OUTPUT)
public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(s -> s.toUpperCase());
}
}
使用输出参数的同一个处理器如下所示：
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<String> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(s -> s.toUpperCase()));
}
}
RxJava 1.x支持
RxJava 1.x处理程序遵循与基于反应器的规则相同的规则，但将使用Observable和ObservableSender参数和返回类型。
所以上面的第一个例子会变成：
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {
@StreamListener
@Output(Processor.OUTPUT)
public Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(s -> s.toUpperCase());
}
}
上面的第二个例子将会变成：
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<String> input,
@Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(s -> s.toUpperCase()));
}
}
聚合
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等）之一。
