使用 Spring Cloud Stream 构建简单的事件驱动微服务

工程 | Ben Wilcock | 2019年10月15日 | ...

事件驱动架构非常棒。但是如果没有框架，编写与流行事件消息平台配合使用的脚手架可能会很混乱。在这篇文章中，我们将探讨如何使用 Spring Cloud Stream 来简化您的代码。

问题

您只想为事件驱动应用程序编写逻辑，但样板消息代码可能会阻碍您。将您的应用程序连接到消息服务很棘手，如果您是企业开发人员，您可能需要使用多种消息技术（无论是本地的还是云端的）。

解决方案

让灵活的消息抽象来处理复杂的 मैसेज平台集成，这样您就可以专注于编写简单清晰的业务逻辑。Spring Cloud Stream 是一个很好的选择。它将许多流行的消息平台统一到一个易于使用的 API 后面，包括 RabbitMQ、Apache Kafka、Amazon Kinesis、Google PubSub、Solace PubSub+、Azure Event Hubs 和 Apache RocketMQ。它甚至可以消除这些平台之间（例如分区或交换）方法和功能上的细微差异，让您可以自由地创建创新的事件驱动解决方案。

在接下来的演示中，您将看到 Spring Cloud Stream 巧妙的抽象如何帮助使事件流代码更简洁、更易于使用。您还将看到使用 Spring Cloud Stream 的 binding 库在两个不同的消息平台（RabbitMQ 或 Kafka）之间切换是多么简单。

开始之前

这些事件驱动微服务需要在您的 PC 上安装以下最新应用程序¹

Java 8
Docker（我们将在本地运行 RabbitMQ 和 Kafka）
Git（可选）
Bash（假定，尽管其他选项也可以）

运行演示

首先，从 GitHub 克隆代码仓库。为此（如果已安装 Git），请打开新的终端窗口并发出以下命令。如果未安装 Git，请下载并解压此 zip 文件。

git clone https://github.com/benwilcock/spring-cloud-stream-demo.git

检查代码后，您会发现此仓库包含两个微服务。

Loansource 微服务（位于 /loansource 文件夹中）。此微服务充当事件消息的源。这些事件是 Loan 应用程序，类似于您在银行和金融领域会看到的。每笔贷款都有一个“名称”、“金额”和“状态”（最初设置为 PENDING）。
Loancheck 微服务（位于 /loancheck 文件夹中）。此微服务充当 Loan 处理器。它检查哪些贷款是可行的，并将其分类为 APPROVED 或 DECLINED 状态。

要运行演示，请按照以下说明操作。

步骤 1：启动消息服务器

在一个新的终端窗口中，进入项目的根文件夹并发出以下命令。

您需要安装并运行 "Docker"，此脚本才能正常工作，因为它需要 docker-compose。

./start-servers.sh

此脚本将启动 Kafka 和 RabbitMQ 并将两者的日志输出流式传输到终端窗口（除非您使用 Ctrl-C 退出）。当您按下 Ctrl-C 时，服务器不会停止 - 它们将在后台继续运行。一旦启动，这些服务器将全部可供您的计算机上运行的应用程序使用。

步骤 2：选择 Kafka 或 RabbitMQ 模式

在接下来的步骤 3 和 4 中，我们必须将 -P<profile-choice> 替换为我们想要使用的消息平台的名称。

对于 Kafka，使用：-Pkafka
对于 RabbitMQ，使用：-Prabbit

如果您完全省略 -P<profile-choice> 设置，则使用 Kafka。

注意：此演示并非旨在在 Kafka 和 RabbitMQ 之间“桥接”消息，因此请确保在编译和运行这两个应用程序时选择相同的配置文件名。如果您的目标是桥接消息系统，请参见此处文档。

步骤 3：生成一些贷款事件

在一个新的终端窗口中，使用 cd 命令将 /loansource 目录设置为当前目录，然后发出以下命令，将 <profile-choice> 替换为您想要运行的模式（如上面步骤 2 中所述，可以是 kafka 或 rabbit 模式）。

./mvnw clean package spring-boot:run -DskipTests=true -P<profile-choice>

一旦 loansource 应用程序启动，在终端窗口中，您应该每秒看到一条消息，告诉您一个新的贷款事件已以 PENDING 状态发布到消息平台。让此微服务继续运行，然后进入下一步。

步骤 4：处理贷款事件

在另一个新的终端窗口中，将 /loancheck 目录设置为当前目录，然后发出以下命令，同样将 <profile-choice> 替换为您想要运行的模式。

./mvnw clean package spring-boot:run -DskipTests=true -P<profile-choice>

一旦 loancheck 应用程序启动，在终端窗口中，您应该每秒看到一条消息，告诉您一个新的 PENDING 贷款申请已从消息平台读取并被 APPROVED 或 DECLINED。如果您想了解这些应用程序是如何构建的，请跳到“工作原理”部分。

步骤 5：停止演示

完成微服务后，在 /loansource 和 /loancheck 微服务的每个终端窗口中按下 Ctrl-C。应用程序将停止，事件处理也将停止。

如果您要在 Kafka 和 Rabbit 之间切换模式，只需返回到步骤 2 并重复该过程。

如果您已完全完成演示并希望停止 Kafka 和 RabbitMQ 服务器，请在项目根文件夹的终端窗口中运行 ./stop-servers.sh 脚本。如果您只是在模式之间切换，则没有必要。

工作原理

Maven 配置文件（在每个项目的 pom.xml 中）控制在构建时添加哪些 Spring Cloud Stream 绑定作为依赖项。当您选择 -Pkafka 时，[spring-cloud-stream-binder-kafka][kafka] 依赖项将添加到项目中。当您选择 -Prabbit 时，[spring-cloud-stream-binder-rabbit][rabbit] 依赖项将添加。

<profiles>
    <profile>
        <id>kafka</id>
        <properties>
            <spring.profile.activated>kafka</spring.profile.activated>
        </properties>
        <activation>
            <activeByDefault>true</activeByDefault>
        </activation>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream-binder-kafka</artifactId>
                <version>${spring-cloud-stream.version}</version>
            </dependency>
        </dependencies>
    </profile>
    ...
<profiles>

您选择的 Maven 配置文件还会影响 src/main/resources/application.properties 文件中的 spring.profiles.active 属性，该属性会切换您在启动时看到的横幅。

贷款源微服务

对于 Loansource 微服务，我们使用了 Spring Cloud Stream v2.1 的一项新功能 - Spring Cloud Function 支持。有了这项新功能，让 LoansourceApplication 微服务充当 Loan 消息的来源所需的全部是声明一个生成并返回 Supplier<> 的 @Bean 方法。在这种情况下，它是一个 Loan 类型的 Supplier。函数方法代码看起来像这样...

@Bean
public Supplier<Loan> supplyLoan() {
  return () -> {
    Loan loan = new Loan(UUID.randomUUID().toString(), "Ben", 10000L);
    LOG.info("{} {} for ${} for {}", loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName());
    return loan;
  };
}

Supplier<> 是 Java 函数数据类型。由于只有一个返回此类型的 @Bean 方法，Spring Cloud Stream 确切地知道接下来该怎么做。默认情况下，它将每秒触发一次此函数并将结果发送到名为“output”的默认 MessageChannel。此函数方法的优点是它只包含业务逻辑，因此您可以使用常规单元测试对其进行测试。

我们可以使用 application.properties 文件中的 spring.cloud.function.definition 属性来明确声明我们希望将哪个函数 bean 绑定到绑定目的地 - 但对于只定义一个 @Bean 的情况，这没有必要。

如果我们想使用不同的轮询间隔，我们可以使用 application.properties 文件中的 spring.integration.poller.fixed-delay 属性。

贷款检查微服务

loancheck 微服务需要更多的代码，但不多。它的工作是将 Loan 事件分类到不同的通道中。为此，它正在订阅来自源的 output 主题的事件，然后根据贷款的价值将其发送到 approved 或 declined 主题，类似于欺诈检查设施。

因为我们使用了 3 个消息通道（一个入站和两个出站），所以使用了一个简单的 LoanProcessor 接口来阐明输入和输出。目前，它看起来像这样

@Component
public interface LoanProcessor {

  String APPLICATIONS_IN = "output"; // Topic where the new loans appear
  String APPROVED_OUT = "approved"; // Topic where the approved loans are sent
  String DECLINED_OUT = "declined"; // Topic where the declined loans are sent

  @Input(APPLICATIONS_IN)
  SubscribableChannel sourceOfLoanApplications();

  @Output(APPROVED_OUT)
  MessageChannel approved();

  @Output(DECLINED_OUT)
  MessageChannel declined();
}

此 LoanProcessor 接口首先在 @SpringBootApplication 类 (LoanCheckApplication.java) 中作为 @EnableBinding() 注解的参数引用，如下所示。

@SpringBootApplication
@EnableBinding(LoanProcessor.class)
public class LoanCheckApplication {

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

此外，一个名为 LoanChecker.java 的 Spring @Component 在运行时使用此 LoanProcessor 构建。此外，此组件的 checkAndSortLoans(Loan) 方法在每次新 Loan 事件到达时都会自动调用，因为它已被注释为 LoanProcessor.APPLICATIONS_IN 通道的 @StreamListener()。您可以在以下代码示例中看到此注解的使用。

  @StreamListener(LoanProcessor.APPLICATIONS_IN)
  public void checkAndSortLoans(Loan loan) {

    if (loan.getAmount() > MAX_AMOUNT) {
      loan.setStatus(Statuses.DECLINED.name());
      processor.declined().send(message(loan));
    } else {
      loan.setStatus(Statuses.APPROVED.name());
      processor.approved().send(message(loan));
    }
  }

此方法随后使用简单的业务逻辑对 Loan 对象进行排序。根据排序结果，它将它们发送到 processor.approved() 通道或 processor.declined() 通道（在相应设置其贷款状态之后）。

总结

正如您所看到的，使用 Spring Cloud Streams 获得的关注点分离确实非常健康。在任何一个微服务中，都没有任何特定于 Kafka 或 RabbitMQ 的代码。这使我们能够专注于业务逻辑，而不管消息平台是什么，并且您可以通过简单地更改项目 pom.xml 中的“绑定器”依赖项来轻松切换消息平台。

Spring 博客