[翻译]Spring Cloud Stream

原文在这里

1.简介

Spring Cloud Stream是创建消息驱动微服务应用的框架。Spring Cloud Stream是基于spring boot创建，用来建立单独的／工业级spring应用，使用spring integration提供与消息代理之间的连接。本文提供不同代理中的中间件配置，介绍了持久化发布订阅机制，以及消费组以及分割的概念。

将注解@EnableBinding加到应用上就可以实现与消息代理的连接，@StreamListener注解加到方法上，使之可以接收处理流的事件。

@SpringBootApplication
public class StreamApplication {

public static void main(String[] args) {
SpringApplication.run(StreamApplication.class, args);
}
}

@EnableBinding(Sink.class)
public class TimerSource {

...

@StreamListener(Sink.INPUT)
public void processVote(Vote vote) {
votingService.recordVote(vote);
}
}

@EnableBinding注解使用一个或者多个接口作为参数（本例子中，参数是单独的Sink接口）。接口声明了输入和／或输出通道。 Spring Cloud Stream提供了Source, Sink, 和 Processor三个接口；你也可以定义你自己的接口。

下面是Sink接口的定义：

public interface Sink {
String INPUT = "input";

@Input(Sink.INPUT)
SubscribableChannel input();
}

@Input注解定义了一个输入通道，应用通过该输入通道接收进入应用的消息；@Output注解定义了一个输出通道，发布的消息通过该通道离开应用。input和output注解可以使用通道名称作为参数；如果没有名称，会使用带注解的方法的名字作为参数（也就是说，如果没有定义单独的名字，这里的通道名就是方法名input）。

Spring Cloud Stream会为你创建一个接口的实现(这里注意，一定要在application里面加上@EnableBinding注解，不然会出现自动注入失败，因为缺少这个注解的话stream就不会创建接口的实例)。你可以通过自动装配在应用中使用它，如下面测试用例所示：

@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = StreamApplication.class)
@WebAppConfiguration
@DirtiesContext
public class StreamApplicationTests {

@Autowired
private Sink sink;

@Test
public void contextLoads() {
assertNotNull(this.sink.input());
}
}

2.主要概念

Spring Cloud Stream提供了很多抽象和基础组件来简化消息驱动型微服务应用。包含以下内容：

Spring Cloud Stream的应用模型

绑定抽象

持久化发布／订阅支持

消费者组支持

分片支持（Partitioning Support）

可插拔绑定api 应用模型

2.1应用模型

Spring Cloud Stream由一个中间件中立的核组成。应用通过Spring Cloud Stream插入的input和output通道与外界交流。通道通过指定中间件的Binder实现与外部代理连接。



2.1.1 胖JAR

测试的话，Spring Cloud Stream可以在ide中运行一个单独的实例。在生产环境中，可以通过Maven 或者 Gradle提供的Spring Boot 工具创建可执行JAR（或者胖JAR）。

2.2绑定抽象

Spring Cloud Stream提供对 Kafka, Rabbit MQ,Redis, 和 Gemfire的绑定实现。

Spring Cloud Stream使用Spring Boot做配置，绑定抽象使得stream可以灵活的连接到中间件。比如，开发者可以在运行时动态的选择通道连接的目标（可以是kafka主题或者RabbitMQ 交换机）。该配置可以通过spring boot支持的任何配置形式实现。在sink的例子中，将属性spring.cloud.stream.bindings.input.destination设置为raw-sensor-data，程序会从命名为raw-sensor-data的kafka主题中读取数据，或者从一个绑定到raw-sensor-data的rabbitmq交换机的队列中读取数据（这里的input是通道名，raw-sensor-data则是exchange的名字，通过使用同一个名字，可以将输入输出通道进行绑定）。

Spring Cloud Stream自动检测和使用在class path中找到的binder。可以在build的时候引入不同的binder来使用不同类型的中间件。更复杂的情况下，可以在应用中打包多个binder使之按需选择，甚至可以在运行时根据不同的通道选择不同的binder。

2.3持久化发布／订阅支持

应用间通信遵照发布-订阅模型，消息通过共享主题进行广播。下图所示，显示了交互的Spring Cloud Stream 应用的典型布局。



sensor传给http端点的数据传给名为raw-sensor-data的目标。发布订阅模型简化了生产者和消费者的复杂程度，并且新的应用可以在不对当前数据流造成影响的情况下加入到拓扑中。

由于发布－订阅模型并非一个新的概念，Spring Cloud Stream将其作为应用模型中的可选项。通过使用原生中间件支持，Spring Cloud Stream也简化了不同平台之间使用发布－订阅模型的复杂程度。

2.4消费者组

由于发布－订阅模型使得共享主题的应用之间连接更简便，创建给定应用的不同实例来进行弹性扩张的能力也同样重要。如果存在多个应用实例，那么同一应用的额不同实例便会成为相互竞争的消费者，其中应该只有一个实例处理给定消息。

Spring Cloud Stream通过消费者组的概念给这种情况进行建模。每一个单独的消费者可以使用spring.cloud.stream.bindings.input.group属性来指定一个组名字。下图中展示的消费者们，这一属性被设置为spring.cloud.stream.bindings.input.group=hdfsWrite或者spring.cloud.stream.bindings.input.group=average。



所有订阅给定目标的组都会收到发布消息的一个拷贝，但是每一个组内只有一个成员会收到该消息。默认情况下，如果没有指定组，Spring Cloud Stream 会将该应用指定给一个匿名的独立的单成员消费者组，后者与所有其他组都处于一个发布－订阅关系中。

持久性

与Spring Cloud Stream中的可选应用模型一样，消费者组订阅是持久的。也就是说，一个绑定的实现确保组的订阅者是持久的，一旦组中至少有一个成员创建了订阅，这个组就会收到消息，即使组中所有的应用都被停止了，组仍然会收到消息。

注：自然情况下，匿名订阅者是非持久化的。对于某些绑定实现（如rabbitmq），可以创建非持久化（non－durable）组订阅。

一般来说，将应用绑定到给定目标的时候，最好指定一个消费者组。扩展Spring Cloud Stream应用的时候，对于它的每一个输入绑定，都必须要指定一个消费者组。 这样可以防止应用实例收到重复的消息。（除非存在重复收到的需求，但实际上很少会有这样的需求）。

2.5分片支持（Partitioning Support）

Spring Cloud Stream对给定应用的多个实例之间分隔数据予以支持。在分隔方案中，物理交流媒介（如：代理主题）被视为分隔成了多个片（partitions）。一个或者多个生产者应用实例给多个消费者应用实例发送消息并确保相同特征的数据被同一消费者实例处理。

Spring Cloud Stream对分割的进程实例实现进行了抽象。因此分片可以用于自带分隔的代理（如kafka）或者不带分隔的代理（如rabbiemq）。



分割在有状态处理中是一个很重要的概念，在性能和一致性上，分割都是重要的概念。例如，在时间窗平均计算的例子中，给定传感器测量结果应该都由同一应用实例进行计算。

注：如果要设置分割处理方案，需要配置数据处理和数据消费端点。

3.编程模型

这一部分描述Spring Cloud Stream的编程模型。Spring Cloud Stream提供很多预先定一的注解来声明约束输入和输出通道以及如何监听这些通道。

3.1声明和绑定通道

3.1.1通过@EnableBinding触发绑定

将@EnableBinding应用到spring应用的一个配置类中，可以将spring应用变成Spring Cloud Stream应用。@EnableBinding注解本身就包含@Configuration注解，并且会触发Spring Cloud Stream 基本配置。

@EnableBinding注解可以接收一个或多个接口类作为对象，后者包含代表了可绑定构件（一般来说是消息通道）的方法。

注：在Spring Cloud Stream1.0中，仅有的可绑定构件是Spring 消息 MessageChannel以及它的扩展SubscribableChannel 和 PollableChannel. 未来版本会使用相同的机制扩展对其他类型构件的支持。在本文档中，会继续饮用通道。

3.1.2@Input 和 @Output

一个Spring Cloud Stream应用可以有任意数目的input和output通道，后者通过@Input and @Output方法在进口中定义。

public interface Barista {

@Input
SubscribableChannel orders();

@Output
MessageChannel hotDrinks();

@Output
MessageChannel coldDrinks();
}

将该接口作为@EnableBinding的参数，会相应的触发三个名为orders, hotDrinks, 和 coldDrinks的绑定好的通道。

@EnableBinding(Barista.class)
public class CafeConfiguration {

...
}

定制通道名字

使用@Input 和 @Output注解，可以自己指定通道的名字，如下所示：

public interface Barista {
...
@Input("inboundOrders")
SubscribableChannel orders();
}

这个例子中，创建的绑定队列会被命名为inboundOrders。

Source, Sink, and Processor

在大多数用例中，包含一个输入通道或者一个输出通道或者二者都包含，为了更简单的定位，Spring Cloud Stream创造性的提供了三个预定义的接口。

Source用于有单个输出（outbound）通道的应用。

public interface Source {

String OUTPUT = "output";

@Output(Source.OUTPUT)
MessageChannel output();

}

Sink用于有单个输入（inbound）通道的应用。

public interface Sink {

String INPUT = "input";

@Input(Sink.INPUT)
SubscribableChannel input();

}

Processor用于单个应用同时包含输入和输出通道的情况。

public interface Processor extends Source, Sink {
}

Spring Cloud Stream对这三个接口没有提供任何特殊处理。他们只是用于创造性的提供。

3.1.3访问绑定通道

1.注入已绑定接口

对于每一个已绑定的接口， Spring Cloud Stream会生成一个bean实现该接口。唤起这些由@Input或者 @Output注解的方法生成的bean，其中一个bean会返回相应的通道。

下面例子中，当hello方法被唤起的时候，bean会在output通道上发送一个消息。在注入的Source bean上提供唤醒output()来检索到目标通道。

@Component
public class SendingBean {

private Source source;

@Autowired
public SendingBean(Source source) {
this.source = source;
}

public void sayHello(String name) {
source.output().send(MessageBuilder.withPayload(body).build());
}
}

2.直接注入到通道

绑定的通道也可以直接注入。

@Component
public class SendingBean {

private MessageChannel output;

@Autowired
public SendingBean(MessageChannel output) {
this.output = output;
}

public void sayHello(String name) {
output.send(MessageBuilder.withPayload(body).build());
}
}

如果通道名称是在声明的注解上指定的，则不能使用方法名称，而要使用通道名称。举例如下：

public interface CustomSource {
...
@Output("customOutput")
MessageChannel output();
}

通道会按照下面方式注入：

@Component
public class SendingBean {

@Autowired
private MessageChannel output;

@Autowired @Qualifier("customOutput")
public SendingBean(MessageChannel output) {
this.output = output;
}

public void sayHello(String name) {
customOutput.send(MessageBuilder.withPayload(body).build());
}
}

3.1.4生产和消费消息

可以使用Spring Integration 的注解或者Spring Cloud Stream的 @StreamListener 注解来实现一个Spring Cloud Stream应用。@StreamListener注解模仿其他spring消息注解（例如@MessageMapping, @JmsListener, @RabbitListener等），但是它增加了内容类型管理和类型强制特性。

1.原生Spring Integration支持

因为 Spring Cloud Stream是基于Spring Integration构建，Stream完全继承了Integration的基础设施以及构件本身。例如，可以将Source的output通道连接到一个MessageSource：

@EnableBinding(Source.class)
public class TimerSource {

@Value("${format}")
private String format;

@Bean
@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
}
}

或者可以在transformer中使用处理器的通道。

@EnableBinding(Processor.class)
public class TransformProcessor {
@Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public Object transform(String message) {
return message.toUpper();
}
}

2.使用 @StreamListener进行自动内容类型处理

作为原生Spring Integration的补充，Spring Cloud Stream提供了自己的@StreamListener注解，该注解模仿spring的其它消息注解（如@MessageMapping, @JmsListener, @RabbitListener等）。@StreamListener注解提供了一种更简单的模型来处理输入消息，尤其是处理包含内容类型管理和类型强制的用例的情况。

Spring Cloud Stream提供了一个扩展的MessageConverter机制，该机制提供绑定通道实现数据处理，本例子中，数据会分发给带@StreamListener注解的方法。下面例子展示了处理外部Vote事件的应用：

@EnableBinding(Sink.class)
public class VoteHandler {

@Autowired
VotingService votingService;

@StreamListener(Sink.INPUT)
public void handle(Vote vote) {
votingService.record(vote);
}
}

在输入消息内容头为application/json的情况下，@StreamListener和Spring Integration的@ServiceActivator之间会体现出差异。使用@StreamListener的情况下，MessageConverter机制会使用contentType头将string负载解析为Vote对象（也就是如果传输的是对象，应该选用@StreamListener注解）。

在其它Spring Messaging方法中，消息机制可以使用@Payload, @Headers 和 @Header这些注解。

注：对于那些有返回数据的方法，必须使用@SendTo注解来指定返回数据的输出绑定目标。

@EnableBinding(Processor.class)
public class TransformProcessor {

@Autowired
VotingService votingService;

@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public VoteResult handle(Vote vote) {
return votingService.record(vote);
}
}

注：在RabbitMQ中，内容类型头可以由外部应用设定。

3.1.5聚合Aggregation

Spring Cloud Stream可以支持多种应用的聚合，可以实现多种应用输入输出通道直接连接，而无需额外代价。在1.0版本中，只有以下类型应用支持聚合：

sources：带有名为output的单一输出通道的应用。典型情况下，该应用带有包含一个以下类型的绑定

org.springframework.cloud.stream.messaging.Source

sinks：带有名为input的单一输入通道的应用。典型情况下，该应用带有包含一个以下类型的绑定

org.springframework.cloud.stream.messaging.Sink

processors：带有名为input的单一输入通道和带有名为output的单一输出通道的应用。典型情况下，该应用带有包含一个以下类型的绑定

type org.springframework.cloud.stream.messaging.Processor.

可以通过创建一系列相互连接的应用将它们聚合到一起，其中，序列中一个元素的输出通道与下一个元素的输入通道连接在一起。序列可以由一个cource或者一个processor开始，可以包含任意数目的processors，并由processors或者sink结束。

取决于开始和结束元素的特性，序列可以有一个或者多个可绑定的通道，如下：

如果序列由source开始，sink结束，应用之间直接通信并且不会绑定通道

如果序列由processor开始，它的输入通道会变成聚合的input通道并进行相应的绑定

如果序列由processor结束，它的输出通道会变成聚合的output通道并进行相应的绑定

使用AggregateApplicationBuilder功能类来实现聚合，如下例子所示。考虑一个包含source,processor 和 sink的工程，它们可以示包含在工程中，或者包含在工程的依赖中。

@SpringBootApplication
@EnableBinding(Sink.class)
public class SinkApplication {

private static Logger logger = LoggerFactory.getLogger(SinkModuleDefinition.class);

@ServiceActivator(inputChannel=Sink.INPUT)
public void loggerSink(Object payload) {
logger.info("Received: " + payload);
}
}

@SpringBootApplication
@EnableBinding(Processor.class)
public class ProcessorApplication {

@Transformer
public String loggerSink(String payload) {
return payload.toUpperCase();
}
}

@SpringBootApplication
@EnableBinding(Source.class)
public class SourceApplication {

@Bean
@InboundChannelAdapter(value = Source.OUTPUT)
public String timerMessageSource() {
return new SimpleDateFormat().format(new Date());
}
}

每一个配置可用于运行一个独立的组件，在这个例子中，它们可以这样实现聚合：

@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()方法的参数。同一类型的多个processors可以链在一起（例如，可以使用不同配置的管道传输方式）。对于每一个组件，编译器可以为Spring Boot 提供运行时参数。

3.1.6RxJava支持

4.绑定器（Binders）

Spring Cloud Stream提供绑定抽象用于与外部中间件中的物理目标进行连接。本章主要介绍Binder SPI背后的主要概念，主要组件以及实现细节。

4.1生产者和消费者

任何往通道中发布消息的组件都可称作生产者。通道可以通过代理的Binder实现与外部消息代理进行绑定。调用bindProducer()方法，第一个参数是代理名称，第二个参数是生产者向其中发送消息的本地通道目标名称，第三个参数包含通道创建的适配器的属性信息（比如：分片key

表达式）。

任何从通道中接收消息的组件都可称作消费者。与生产者一起，消费者通道可以与外部消息代理进行绑定。调用bindConsumer()方法，第一个参数是目标名称，第二个参数提供了消费者逻辑组的名称。

4.2Binder SPI

Binder SPI 由若干接口组成。

绑定指定配置Binder-Specific Configuration

以下绑定器／消费者和生产者属性是指定给binder 实现的。

Rabbit-Specific Settings