Spring 数据集成之旅的简史
Spring 的数据集成之旅始于 Spring Integration。凭借其编程模型，它提供了始终如一的开发体验，用于构建能够采用 企业集成模式 的应用程序，以连接数据库、消息代理等外部系统。 快进到云时代，微服务在企业环境中变得突出。Spring Boot 改变了开发人员构建应用程序的方式。凭借 Spring 的编程模型和 Spring Boot 处理的运行时职责，开发独立的、生产级的基于 Spring 的微服务变得无缝。 为了将其扩展到数据集成工作负载，Spring Integration 和 Spring Boot 被整合到一个新项目中。Spring Cloud Stream 应运而生。 通过 Spring Cloud Stream，开发人员可以：
-
独立构建、测试和部署以数据为中心的应用程序。
-
应用现代微服务架构模式，包括通过消息传递进行组合。
-
通过以事件为中心的思维解耦应用程序职责。事件可以表示在特定时间发生的事情，下游消费者应用程序可以对其做出反应，而无需知道事件的来源或生产者的身份。
-
将业务逻辑移植到消息代理（例如 RabbitMQ、Apache Kafka、Amazon Kinesis）上。
-
依赖框架对常见用例的自动内容类型支持。扩展到不同的数据转换类型是可能的。
-
还有更多。 . .
快速入门
即使在深入了解任何细节之前，您也可以按照这个三步指南，在不到 5 分钟的时间内尝试 Spring Cloud Stream。
我们将向您展示如何创建一个 Spring Cloud Stream 应用程序，该应用程序接收来自您选择的消息中间件（稍后会详细介绍）的消息，并将接收到的消息记录到控制台。
我们称之为 LoggingConsumer。
虽然不切实际，但它很好地介绍了核心概念和抽象，使消化本用户指南的其余部分变得更容易。
这三个步骤如下：
使用 Spring Initializr 创建示例应用程序
首先，访问 Spring Initializr。在那里，您可以生成我们的 LoggingConsumer 应用程序。为此：
-
在 Dependencies 部分，开始输入
stream。 当出现“Cloud Stream”选项时，选择它。 -
开始输入“kafka”或“rabbit”。
-
选择“
Kafka”或“RabbitMQ”。基本上，您选择应用程序绑定的消息中间件。 我们建议使用您已安装或更熟悉安装和运行的那个。 此外，正如您从 Initializr 屏幕上看到的，还有其他一些选项可供选择。 例如，您可以选择 Gradle 作为构建工具而不是 Maven（默认）。 -
在 Artifact 字段中，输入“logging-consumer”。Artifact 字段的值将成为应用程序名称。 如果您选择 RabbitMQ 作为中间件，您的 Spring Initializr 现在应该如下所示：
-
点击 Generate Project 按钮。这样做会将生成的项目的压缩版本下载到您的硬盘驱动器。
-
将文件解压缩到您想要用作项目目录的文件夹中。
|
我们鼓励您探索 Spring Initializr 中提供的许多可能性。 它允许您创建许多不同类型的 Spring 应用程序。 |
将项目导入到您的 IDE 中
现在您可以将项目导入到您的 IDE 中。 请记住，根据 IDE 的不同，您可能需要遵循特定的导入过程。 例如，根据项目生成方式（Maven 或 Gradle），您可能需要遵循特定的导入过程（例如，在 Eclipse 或 STS 中，您需要使用 File → Import → Maven → Existing Maven Project）。
导入后，项目不得有任何错误。此外，src/main/java 应该包含 com.example.loggingconsumer.LoggingConsumerApplication。
从技术上讲，此时您可以运行应用程序的主类。 它已经是一个有效的 Spring Boot 应用程序。 但是，它不做任何事情，所以我们想添加一些代码。
添加消息处理器、构建和运行
修改 com.example.loggingconsumer.LoggingConsumerApplication 类，使其如下所示：
@SpringBootApplication
public class LoggingConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(LoggingConsumerApplication.class, args);
}
@Bean
public Consumer<Person> log() {
return person -> {
System.out.println("Received: " + person);
};
}
public static class Person {
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String toString() {
return this.name;
}
}
}正如您从前面的列表中看到的：
-
我们正在使用函数式编程模型（参见 [Spring Cloud Function support]）将单个消息处理器定义为
Consumer。 -
我们依赖框架约定将此类处理器绑定到由绑定器公开的输入目标绑定。
这样做还可以让您看到框架的一个核心功能：它尝试自动将传入消息负载转换为 Person 类型。
您现在拥有一个功能齐全的 Spring Cloud Stream 应用程序，它侦听消息。
从这里，为简单起见，我们假设您在 spring-cloud-stream-preface-creating-sample-application 中选择了 RabbitMQ。
假设您已安装并运行 RabbitMQ，您可以通过在 IDE 中运行其 main 方法来启动应用程序。
您应该看到以下输出：
--- [ main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg, bound to: input
--- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [localhost:5672]
--- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#2a3a299:0/SimpleConnection@66c83fc8. . .
. . .
--- [ main] o.s.i.a.i.AmqpInboundChannelAdapter : started inbound.input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg
. . .
--- [ main] c.e.l.LoggingConsumerApplication : Started LoggingConsumerApplication in 2.531 seconds (JVM running for 2.897)转到 RabbitMQ 管理控制台或任何其他 RabbitMQ 客户端，并将消息发送到 input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg。
anonymous.CbMIwdkJSBO1ZoPDOtHtCg 部分表示组名，它是生成的，因此在您的环境中必然不同。
为了更具可预测性，您可以通过设置 spring.cloud.stream.bindings.input.group=hello（或任何您喜欢的名称）来使用显式组名。
消息的内容应该是 Person 类的 JSON 表示，如下所示：
{"name":"Sam Spade"}
然后，在您的控制台中，您应该看到：
Received: Sam Spade
您还可以将应用程序构建并打包成一个 boot jar（使用 ./mvnw clean install），然后使用 java -jar 命令运行构建的 JAR。
现在您有了一个正常工作的（尽管非常基本）Spring Cloud Stream 应用程序。
流式数据上下文中的 Spring 表达式语言 (SpEL)
在本参考手册中，您将遇到许多可以利用 Spring 表达式语言 (SpEL) 的特性和示例。了解使用它时某些限制非常重要。
SpEL 允许您访问当前消息以及您正在运行的应用程序上下文。
但是，了解 SpEL 可以看到的数据类型，尤其是在传入消息的上下文中，非常重要。
从代理接收到的消息以 byte[] 的形式出现。然后，绑定器将其转换为 Message<byte[]>，其中您可以看到消息的负载保持其原始形式。消息的标头是 <String, Object>，其中值通常是另一个基本类型或基本类型的集合/数组，因此是 Object。
这是因为绑定器不知道所需的输入类型，因为它无法访问用户代码（函数）。因此，绑定器有效地传递了一个包含负载和一些可读元数据（以消息标头形式）的信封，就像邮件发送的信件一样。
这意味着，虽然可以访问消息的负载，但您只能将其作为原始数据（即 byte[]）访问。虽然开发人员通常会要求 SpEL 能够访问负载对象的字段作为具体类型（例如 Foo、Bar 等），但您可以看出实现起来是多么困难甚至不可能。
这里有一个例子来演示这个问题；想象一下您有一个路由表达式，根据负载类型路由到不同的函数。此要求意味着将负载从 byte[] 转换为特定类型，然后应用 SpEL。但是，为了执行此类转换，我们需要知道要传递给转换器的实际类型，而这来自我们不知道的函数签名。解决此要求的更好方法是将类型信息作为消息标头传递（例如，application/json;type=foo.bar.Baz）。您将获得一个清晰可读的字符串值，该值可以在一年内轻松访问和评估，并且易于阅读的 SpEL 表达式。
此外，使用负载进行路由决策被认为是非常糟糕的做法，因为负载被认为是特权数据——只有最终接收者才能读取的数据。同样，使用邮件投递类比，您不会希望邮递员打开您的信封并阅读信件内容以做出一些投递决定。相同的概念也适用于此，尤其是当在生成消息时包含此类信息相对容易时。它强制执行与网络传输数据设计相关的某些级别的纪律，以及此类数据的哪些部分可以被视为公共的，哪些是特权数据。