概述

流式微服务主要用于实时处理源源不断的数据流，相对于应用微服务，它对开发人员提出了更多的技术要求。数据本身不像是纯业务应用，相对来说它是抽象的、复杂的，且在传输，存储，分析等方面都比应用微服务提出了更高的性能要求。

在设计上，Spring Cloud Stream 是一个用于构建消息驱动的微服务框架，它基于Spring Boot来创建对 DevOps 提供良好支持的微服务应用，通过 Spring Integration 集成工具为消息连接与传播提供一个统一的、灵活的操作平台。该操作平台是可配置的，比如配置 生产者/消费者 、消费者所属消费者组甚至是消息分区（需要消息中间件提供商本身的支持）。

Spring Cloud Stream的使用非常简单，在应用的main启动类上加上 @EnableBinding 注解就可以连接消息中间件，创建一个流式微服务。通过在方法上添加 @StreamListener 注解就可以异步处理所接收到的消息。

快速开始

下面通过一个简单的 sink 应用来演示Stream应用的消息接收者用法。

@SpringBootApplication @EnableBinding(Sink.class) public class VoteRecordingSinkApplication {    public static void main(String[] args) {     SpringApplication.run(VoteRecordingSinkApplication.class, args);   }    @StreamListener(Sink.INPUT)   public void processVote(Vote vote) {       votingService.recordVote(vote);   } } 

@EnableBinding注解可以接收一个或多个输入/输出通道接口类作为参数(本示例中注解的参数只有 Sink 一个接口)，Spring Cloud Task默认提供了 Source, Sink, 和 Processor接口来定义消息源，接收者和处理通道。通道支持之定义扩展。

下面是 Sink 消息接收者通道的接口定义代码。

public interface Sink {   String INPUT = "input";    @Input(Sink.INPUT)   SubscribableChannel input(); } 

@Input注解定义了输入通道，通过这个注解可以接收进入应用的消息。@Output注解定义了消息输出通道法，通过这个注解可以定义应用发布消息的通道。这两个注解的参数为输入/输出消息通道的名称，如果没有给注解指定参数，默认使用注解所标注的方法的方法名作为通道名称。

Spring Cloud Stream会通过动态代理技术自动为消息通道接口创建实现类。下面代码是测试消息通道是否创建成功的方法。

@RunWith(SpringJUnit4ClassRunner.class) @SpringApplicationConfiguration(classes = VoteRecordingSinkApplication.class) @WebAppConfiguration @DirtiesContext public class StreamApplicationTests {    @Autowired   private Sink sink;    @Test   public void contextLoads() {     assertNotNull(this.sink.input());   } } 

核心概念

Spring Cloud Stream为简化消息驱动微服务设计了许多抽象和基础组件，其核心简要概述如下所示。

• Spring Cloud 流式微服务模型
• Binder 抽象消息中间件
• 持续发布/订阅支持
• 支持消费者组
• 支持消息分区
• 可拔插的 Binder API

1. 流式微服务模型

一个 Spring Cloud Stream 应用以消息中间件为核心，应用通过Spring Cloud Stream注入的输入/输出通道 channels 与外部进行通信。channels 通过特定的Binder实现与外部消息中间件进行通信。如图1Spring Cloud Stream 应用架构示意图所示。

2. 对消息中间件的抽象

Spring Cloud Stream提供了对Kafka和Rabbit MQ的抽象Binder来代表消息中间件，其自身也提供了测试用的中间件 TestSupportBinder，可以直接向通道发送可靠的消息并进行断言测试，据此可以使用扩展API编写自己的Binder。

Spring Cloud Stream使用通用的Spring Boot配置方式，抽象的Binder为灵活配置如何连接消息中间件及发送消息提供了良好的支持。例如，消息发布者可以在运行时动态的选择通道要连接的目标(kafka的topic或 RabbitMQ的exchanges)，这些外部属性配置可以通过Spring Boot给出(通过应用参数，环境变量，application.yml或者application.properties配置文件)。

Spring Cloud Stream会自动发现并使用 classpath 中可用的 绑定器(binder)，你可以通过在构建项目时依赖不同的中间件来使同一份代码支持多种消息中间件。在复杂的应用场景下，你也可以在事先指定好每个通道绑定哪个Binder，直接打包多个binders。

3. 持续发布/订阅支持

流式微服务应用之间通过发布/订阅模型通信，通过共享的话题topic来传播数据。图2展示了流式微服务集合的部署架构图，直观的说明了多个流式微服务之间的关系。

经过HTTP接口的数据被转发到raw-sensor-data目标主题。计算平均时间窗口和持久化数据到HDFS两个相互独立的微服务共同消费这个主题。

这个基于发布订阅的通信模型可以减少生产者和消费者的复杂性，且能够在不破坏现有数据流的情况下扩充拓扑逻辑。比如，你可以在计算平均时间窗口服务后增加一个用于监控和展示最高值的应用，甚至还可以在同一个数据流中添加一个计算平均错误时间的应用。微服务间通过共享主题的方式进行通信比点对点通信的方式大大降低了服务之间的耦合。

发布订阅模型并非新概念，Spring Cloud Stream只是通过一些额外的步骤，使发布订阅模型成为构建微服务的一种极佳选择。通过对本地消息中间提供支持，Spring Cloud Stream可以简洁的构建跨平台的发布订阅模型。

4. 消费者组

发布订阅模型让应用间通过共享话题topic通信连接变得相当容易，不过在为高可用部署多个应用实例时，还需要防止应用对该话题topic中的消息重复消费。多个应用实例之间应该是一个竞争消费的关系，一条消息应该只能被一个消费者消费。

Spring Cloud Stream通过消费者组的概念来模拟上述需求。每个消费者输入通道绑定器可以使用spring.cloud.stream.bindings.<channelName>.group属性来指定其所属消费者组。

4.1 持久的消费者组

与Spring Cloud Stream的可独立运行服务模型一样，消费者组的订阅也是持久的。这句话的意思是说绑定器默认自己自己所属的消费者组是持久存在的，只要该消费者组中的某一个消费者被创建，除非所有的消费者都被停止掉，否则该组将不停的接收消息。

通常情况下，在已知消息目的地的时候，我们都会指定消费者组，在升级到Spring Cloud Stream应用时，你必须要为美俄输入绑定器指定其所属消费者组，这样的话就可以避免在部署多个应用实例时重复消费消息。

分区支持

Spring Cloud Stream为多个生产者实例的应用提供消息分区的支持。在分区的情景下，物理上连接的媒体被视为多分区的结构，一个或多个生产者发送数据到多个消费者，并保证某个具有某些特征的数据仅被某一个消费者实例消费。

Spring Cloud Stream提供公用的抽象，以统一的方式实现分区处理。具体的分区策略可以由提供分区机制的消息中间件来实现。

分区是状态处理中重要的概念。为确保相关数据能够被一个服务一起处理，无论在性能方面还是一致性方面，分区的概念都是非常重要的。

关于

示例源码

spring-cloud-stream-learning 的 stream-simple-listener 子项目

后记

Spring Cloud Stream提供的流式微服务是一个全新的概念，以消息驱动为服务通信方式，异步高性能的进行数据流实时处理。不过其并未采用什么新技术，而是以优美的设计，来抽象各个消息中间件，屏蔽其内部实现原理，使服务间的通信变得更为简单友好。

本文内容主要是对 Spring Cloud Stream Elmhurst 官方文档的翻译，不过作者水平有限，有不尽然的地方敬请指出。本项目和文档中所用的内容仅供学习和研究之用，转载或引用时请指明出处。如果你对文档有疑问或问题，请在项目中给我留言或发email到 weiwei02@vip.qq.com 我的github: https://github.com/weiwei02/ 我相信技术能够改变世界 。

链接

• 上篇文章暂无
• 下篇文章正在准备中