1. 概述

Spring Cloud Stream是一个建立在Spring Boot和Spring Integration之上的框架，有助于创建事件驱动或消息驱动的微服务。

2. Maven 依赖项

首先，需要将 Spring Cloud Starter Stream 与代理 RabbitMQ Maven 依赖项作为消息中间件添加到我们的 pom.xml：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    <version>3.1.3</version>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-test-support</artifactId>
    <version>3.1.3</version>
    <scope>test</scope>
</dependency>

3. 主要概念

微服务架构遵循“smart endpoints and dumb pipes”原则。端点之间的通信由消息中间件方（如RabbitMQ或Apache Kafka）驱动。服务通过这些终结点或通道发布域事件进行通信。

让我们来看看构成 Spring Cloud Stream 框架的概念，以及构建消息驱动服务时必须了解的基本范式。

3.1. 构造

让我们看一下 Spring Cloud Stream中的一个简单的服务，它侦听输入绑定并向输出绑定发送响应：

@SpringBootApplication
@EnableBinding(Processor.class)
public class MyLoggerServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(MyLoggerServiceApplication.class, args);
    }

    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public LogMessage enrichLogMessage(LogMessage log) {
        return new LogMessage(String.format("[1]: %s", log.getMessage()));
    }
}

注解@EnableBinding将应用程序配置为绑定接口处理器中定义的通道 INPUT 和 OUTPUT。这两个通道都是绑定，可以配置为使用具体的消息中间件或绑定器。

让我们来看看所有这些概念的定义：

• Bindings— 以声明方式标识输入和输出通道的接口集合
• Binder — 消息中间件实现，如 Kafka 或 RabbitMQ
• Channel— 表示消息中间件和应用程序之间的通信管道
• StreamListeners — Bean 中的消息处理方法，MessageConverter 在特定于中间件的事件和域对象类型/POJO 之间进行序列化/反序列化后，将自动调用来自通道的消息
• Message Schemas — 用于消息的序列化和反序列化，这些模式可以从某个位置静态读取或动态加载，支持域对象类型的演变

3.2. 通信模式

指定到目标的消息由发布-订阅消息传递模式传递。发布者将邮件分类为主题，每个主题由名称标识。订阅者对一个或多个主题表示兴趣。中间件过滤消息，将有趣的主题传递给订阅者。

现在，订阅者可以分组。使用者组是一组订阅者或使用者，由组 ID 标识，其中来自主题或主题分区的消息以负载平衡的方式传递。

4. 编程模型

本节介绍构建Spring Cloud Stream应用程序的基础知识。

4.1. 功能测试

测试支持是一个绑定程序实现，允许与通道交互并检查消息。让我们向上面的 enrichLogMessage 服务发送一条消息，并检查响应是否在消息开头包含文本 “[1]：”：

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = MyLoggerServiceApplication.class)
@DirtiesContext
public class MyLoggerApplicationTests {

    @Autowired
    private Processor pipe;

    @Autowired
    private MessageCollector messageCollector;

    @Test
    public void whenSendMessage_thenResponseShouldUpdateText() {
        pipe.input()
          .send(MessageBuilder.withPayload(new LogMessage("This is my message"))
          .build());

        Object payload = messageCollector.forChannel(pipe.output())
          .poll()
          .getPayload();

        assertEquals("[1]: This is my message", payload.toString());
    }
}

4.2. 自定义渠道

在上面的例子中，我们使用了Spring Cloud提供的处理器接口，它只有一个输入和一个输出通道。

如果我们需要不同的东西，比如一个输入和两个输出通道，我们可以创建一个自定义处理器：

public interface MyProcessor {
    String INPUT = "myInput";

    @Input
    SubscribableChannel myInput();

    @Output("myOutput")
    MessageChannel anOutput();

    @Output
    MessageChannel anotherOutput();
}

Spring 将为我们提供此接口的正确实现。通道名称可以使用注解进行设置，例如在@Output(“myOutput”)中。否则，Spring 将使用方法名称作为通道名称。因此，我们有三个通道，分别称为myInput，myOutput和Another Output。

现在，假设如果值小于 10，我们希望将消息路由到一个输出，如果值大于或等于 10，则路由到另一个输出：

@Autowired
private MyProcessor processor;

@StreamListener(MyProcessor.INPUT)
public void routeValues(Integer val) {
    if (val < 10) {
        processor.anOutput().send(message(val));
    } else {
        processor.anotherOutput().send(message(val));
    }
}

private static final <T> Message<T> message(T val) {
    return MessageBuilder.withPayload(val).build();
}

4.3. 有条件派遣

使用 @StreamListener 注解，我们还可以使用使用 SpEL 表达式定义的任何条件来过滤我们在消费者中期望的消息。

例如，我们可以使用条件调度作为将消息路由到不同输出的另一种方法：

@Autowired
private MyProcessor processor;

@StreamListener(
  target = MyProcessor.INPUT, 
  condition = "payload < 10")
public void routeValuesToAnOutput(Integer val) {
    processor.anOutput().send(message(val));
}

@StreamListener(
  target = MyProcessor.INPUT, 
  condition = "payload >= 10")
public void routeValuesToAnotherOutput(Integer val) {
    processor.anotherOutput().send(message(val));
}

此方法的唯一限制是这些方法不得返回值。

5. 设置

让我们设置将处理来自 RabbitMQ 代理的消息的应用程序。

5.1. Binder配置

可以将应用程序配置为通过 META-INF/spring.binder 使用默认的绑定器实现：

rabbit:\
org.springframework.cloud.stream.binder.rabbit.config.RabbitMessageChannelBinderConfiguration

或者可以通过包含以下依赖项将 RabbitMQ 的绑定库添加到类路径中：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
    <version>1.3.0.RELEASE</version>
</dependency>

如果没有提供绑定器实现，Spring 将使用通道之间的直接消息通信。

5.2. RabbitMQ配置

要将第 3.1 节中的示例配置为使用 RabbitMQ binder，需要更新位于 src/main/resources 的 application.yml：

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: queue.log.messages
          binder: local_rabbit
        output:
          destination: queue.pretty.log.messages
          binder: local_rabbit
      binders:
        local_rabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: <host>
                port: 5672
                username: <username>
                password: <password>
                virtual-host: /

输入绑定将使用名为 queue.log.messages 的交换，输出绑定将使用交换 queue.pretty.log.messages。两个绑定都将使用名为 local_rabbit 的绑定程序。

请注意，不需要提前创建 RabbitMQ 交换器所或队列。运行应用程序时，将自动创建两个交换。

为了测试应用程序，我们可以使用 RabbitMQ 管理站点发布消息。在交换队列.log.messages的发布消息面板中，我们需要以JSON格式输入请求。

5.3. 自定义消息转换

Spring Cloud Stream允许对特定内容类型应用消息转换。在上面的示例中，希望提供纯文本，而不是使用 JSON 格式。

为此，将使用 MessageConverter 将自定义转换应用于 LogMessage：

@SpringBootApplication
@EnableBinding(Processor.class)
public class MyLoggerServiceApplication {
    //...

    @Bean
    public MessageConverter providesTextPlainMessageConverter() {
        return new TextPlainMessageConverter();
    }

    //...
}

public class TextPlainMessageConverter extends AbstractMessageConverter {

    public TextPlainMessageConverter() {
        super(new MimeType("text", "plain"));
    }

    @Override
    protected boolean supports(Class<?> clazz) {
        return (LogMessage.class == clazz);
    }

    @Override
    protected Object convertFromInternal(Message<?> message, 
        Class<?> targetClass, Object conversionHint) {
        Object payload = message.getPayload();
        String text = payload instanceof String 
          ? (String) payload 
          : new String((byte[]) payload);
        return new LogMessage(text);
    }
}

应用这些更改后，返回到“发布消息”面板，如果将标题“内容类型”设置为“文本/纯文本”，将有效负载设置为“Hello World”，它应该像以前一样工作。

5.4. 消费者Groups

当运行应用程序的多个实例时，每次输入通道中有新消息时，都会通知所有订阅者。大多数情况下，我们只需要处理一次消息。Spring Cloud Stream 通过使用者组实现此行为。

若要启用此行为，每个使用者绑定都可以使用 spring.cloud.stream.bindings.<CHANNEL>.group 属性来指定组名：

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: queue.log.messages
          binder: local_rabbit
          group: logMessageConsumers
          ...

6. 消息驱动的微服务

在本节中，将介绍在微服务上下文中运行Spring Cloud Stream应用程序所需的所有功能。

6.1. 扩大规模

当多个应用程序正在运行时，确保数据在使用者之间正确拆分非常重要。为此，Spring Cloud Stream 提供了两个属性：

• spring.cloud.stream.instanceCount — 正在运行的应用程序数量
• spring.cloud.stream.instanceIndex — 当前应用程序的索引

例如，如果部署了上述 MyLoggerServiceApplication 应用程序的两个实例，则两个应用程序的属性 spring.cloud.stream.instanceCount 应为 2，属性 spring.cloud.stream.instanceIndex 应分别为 0 和 1。

如果按照本文所述使用 Spring 数据流部署 Spring Cloud Stream 应用程序，则会自动设置这些属性。

6.2. 分区

域事件可以是分区消息。当我们扩展存储和提高应用程序性能时，这会有所帮助。

域事件通常具有分区键，以便它最终与相关消息位于同一分区中。

假设我们希望日志消息按消息中的第一个字母（即分区键）进行分区，并分组为两个分区。

以 A-M 开头的日志消息将有一个分区，N-Z 将有一个分区。这可以使用两个属性进行配置：

• spring.cloud.stream.bindings.output.producer.partitionKeyExpression — 用于对有效负载进行分区的表达式
• spring.cloud.stream.bindings.output.producer.partitionCount — 组的数量

有时要分区的表达式太复杂了，无法只用一行编写。对于这些情况，我们可以使用属性spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass编写自定义分区策略。

6.3. 健康指标

在微服务上下文中，还需要检测服务何时关闭或开始失败。Spring Cloud Stream 提供属性管理.health.binders.enabled 以启用活页夹的健康指示器。运行应用程序时，我们可以在 http://<host>:<port>/health 查询健康状态。