admin管理员组

文章数量:1441399

Spring Cloud Stream 简介

一、概述

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

在本文中，我们将通过一些简单的示例来介绍 Spring Cloud Stream 的概念和构造。

2.Maven依赖

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

代码语言：javascript代码运行次数：0运行复制
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    <version>3.1.3</version>
</dependency>

我们还将添加来自 Maven Central 的模块依赖项以启用 JUnit 支持：

代码语言：javascript代码运行次数：0运行复制
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-test-support</artifactId>
    <version>3.1.3</version>
    <scope>test</scope>
</dependency>

三、主要概念

微服务架构遵循“智能端点和哑管道”原则。端点之间的通信由消息中间件方驱动，如 RabbitMQ 或 Apache Kafka。服务通过这些端点或通道发布域事件进行通信。

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

3.1。结构体

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

代码语言：javascript代码运行次数：0运行复制
@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将应用程序配置为绑定在接口Processor中定义的通道INPUT和OUTPUT。两个通道都是绑定，可以配置为使用具体的消息传递中间件或绑定器。

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

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

3.2. 沟通模式

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

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

4. 编程模型

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

4.1。功能测试

测试支持是一个活页夹实现，它允许与通道交互并检查消息。

让我们向上面的enrichLogMessage服务发送一条消息，并检查响应是否在消息的开头包含文本“[1]:”：

代码语言：javascript代码运行次数：0运行复制
@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 提供的Processor接口，它只有一个输入和一个输出通道。

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

代码语言：javascript代码运行次数：0运行复制
public interface MyProcessor {
    String INPUT = "myInput";

    @Input
    SubscribableChannel myInput();

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

    @Output
    MessageChannel anotherOutput();
}

Spring 将为我们提供该接口的正确实现。可以使用@Output(“myOutput”)中的注释设置通道名称。

否则，Spring 将使用方法名称作为通道名称。因此，我们有三个通道，分别称为myInput、myOutput和anotherOutput。

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

代码语言：javascript代码运行次数：0运行复制
@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 表达式定义的任何条件过滤我们期望在消费者中的消息。

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

代码语言：javascript代码运行次数：0运行复制
@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。活页夹配置

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

代码语言：javascript代码运行次数：0运行复制
rabbit:\
org.springframework.cloud.stream.binder.rabbit.config.RabbitMessageChannelBinderConfiguration

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

代码语言：javascript代码运行次数：0运行复制
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
    <version>1.3.0.RELEASE</version>
</dependency>

如果没有提供 binder 实现，Spring 将在通道之间使用直接消息通信。

5.2. RabbitMQ 配置

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

代码语言：javascript代码运行次数：0运行复制
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 管理站点发布消息。在exchange queue.log.messages的Publish Message面板中，我们需要输入 JSON 格式的请求。

5.3. 自定义消息转换

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

为此，我们将使用MessageConverter对LogMessage应用自定义转换：

代码语言：javascript代码运行次数：0运行复制
@SpringBootApplication
@EnableBinding(Processor.class)
public class MyLoggerServiceApplication {
    //...

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

    //...
}
代码语言：javascript代码运行次数：0运行复制
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);
    }
}

应用这些更改后，返回Publish Message面板，如果我们将标题“ contentTypes ”设置为“ text/plain ”，并将有效负载设置为“ Hello World ”，它应该像以前一样工作。

5.4. 消费群体

当运行我们应用程序的多个实例时，每次输入通道中有新消息时，都会通知所有订阅者。

大多数时候，我们只需要处理一次消息。Spring Cloud Stream 通过消费者组实现此行为。

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

代码语言：javascript代码运行次数：0运行复制
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 Data Flow 部署 Spring Cloud Stream 应用程序，这些属性会自动设置。

6.2. 分区

域事件可以是分区消息。这有助于我们扩大存储和提高应用程序性能。

域事件通常有一个分区键，因此它最终与相关消息在同一个分区中。

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

以AM开头的日志消息将有一个分区， NZ将有另一个分区。这可以使用两个属性进行配置：

  • 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 提供了management.health.binders.enabled属性来启用 binder 的健康指标。

运行应用时，我们可以在http://<host>:<port>/health查询健康状态。

本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。 原始发表：2022-05-10，如有侵权请联系 cloudcommunity@tencent 删除springcloudstream配置中间件

本文标签： Spring Cloud Stream 简介

版权声明：本文标题：Spring Cloud Stream 简介 内容由网友自发贡献，该文观点仅代表作者本人， 转载请联系作者并注明出处：http://www.betaflare.com/biancheng/1747918764a2777043.html， 本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容，一经查实，本站将立刻删除。