vlambda博客
学习文章列表

 分布式系统开发的早期阶段，实现location 透明度和弹性是使用外部负载均衡器（例如 HAProxy/Nginx）作为副本组顶部的入口点或作为中央负载均衡器整个系统。考虑下图：

图 8.1 外部服务负载均衡示例

编号图的每个部分解释如下：

从图中我们可以看出，负载 balancer 扮演着可用实例注册的角色。对于每组服务，都有一个专用的负载均衡器来管理在所有实例中的负载。反过来，负载平衡器可以基于整体组负载和可用指标启动扩展过程。例如，当用户活动出现高峰时，可能会动态地将新实例添加到组中，从而处理增加的负载。反过来，当负载减少时，负载均衡器（作为指标持有者）可能会发送通知，说明组中存在冗余实例。

但是，该解决方案存在一些已知问题。首先，在高负载下，负载均衡器可能会成为系统中的热点。回顾 Amdahl 定律，我们可能还记得负载均衡器成为一个争论点，并且底层服务组不能处理比​​负载均衡器更多的请求。反过来，为负载均衡器提供服务的成本可能很高，因为每个专用负载均衡器都需要单独的强大机器或虚拟机。此外，它可能需要安装负载平衡器的额外备份计算机。最后，还应管理和监控负载均衡器。这可能会导致基础设施管理的一些额外费用。

幸运的是，当服务器端负载均衡器成为系统中的热点时，Spring Cloud 生态系统来救援并试图解决这个问题。 Spring 团队没有为外部负载均衡器提供解决方法，而是决定遵循 Netflix 构建分布式系统的最佳实践。实现可扩展性和位置透明性的方法之一是通过客户端负载平衡。

客户端负载平衡的想法很简单，意味着服务通过复杂的客户端进行通信，该客户端知道目标服务的可用实例，以便它可以轻松地平衡它们之间的负载：

图 8.2。客户端负载平衡模式示例

在上图中，编号点的含义如下：

在此示例中，所有调用者执行对 不同 副本的调用 Service A. 尽管该技术提供了独立于专用外部负载平衡器的独立性（从而提供了更好的可扩展性），但它也有其局限性。首先， 提到的负载平衡技术是客户端平衡。因此，每个调用者都负责通过选择目标服务的实例来平衡本地的所有请求。这有助于避免单点故障，从而提供更好的可扩展性。另一方面，关于可用服务的信息应该以某种方式可供系统中的其他服务访问。

为了熟悉用于服务发现的现代技术，我们正在考虑 Java 和 Spring 生态系统中流行的库之一——Ribbon。 Ribbon 库是 Netflix 创建的客户端负载均衡器模式的实现。 Ribbon 提供了两种常用技术来提供对可用服务列表的访问。提供此类信息的最简单方法是通过服务地址的静态预配置列表。该技术如下图所示：

图 8.3。使用 Netflix Ribbon 为每个客户端负载平衡预配置的静态服务列表。

在上述图表中，编号点的含义如下：

在上图中，不同边框的形状 勾勒出不同调用者的知识领域。不幸的是，客户端负载均衡 技术 有其差距。首先，客户端平衡器之间没有协调，所以有可能所有调用者都决定调用同一个instance 并重载它，如下图所示：

图 8.4。不同步的客户端负载均衡问题点的表示

编号图的每个部分解释如下：

此外，这种使用静态服务实例列表管理负载的简单方法远非反应式系统需求，主要来自弹性负载管理 观点。

作为一种解决方案，Ribbon 能够与 Eureka 等服务注册中心集成，因此注册中心服务会不断更新可用服务副本的列表。考虑下图：

图 8.5。可用服务列表的动态更新示例

在上图中，编号点的含义如下：

从上图可以看出，客户端负载均衡器的coordination问题依然存在。 在这种情况下，注册中心负责保存健康服务实例的列表并在运行时不断更新它们。这里，客户端均衡器和注册服务都可以保存目标服务实例的负载信息，客户端均衡器可以周期性地将内部负载统计信息与注册服务收集的负载进行同步。此外，所有相关方都可以访问该信息并根据该信息执行负载平衡。这种管理可用服务列表的方式比前一种更广泛，可以动态更新可用目的地列表。

首先，该技术适用于小型服务集群。 然而，使用共享注册表发现动态服务远非理想。与服务器端负载平衡一样，像 Eureka 这样的经典注册表服务会成为单点故障，因为需要付出巨大的努力来保持系统状态信息的更新和准确。例如，当集群状态 迅速 变化时，注册服务的信息可能会过时。为了跟踪服务的健康状态，服务实例通常会定期发送心跳消息。或者，注册表可以定期执行健康检查请求。在这两种情况下，非常频繁的状态更新可能会消耗不合理的高百分比的集群资源。因此，健康检查之间的持续时间通常从几秒到几分钟不等（Eureka 的默认持续时间是 30 秒）。因此，注册中心可能会提议一个在上次健康检查期间健康但已被销毁的服务实例。因此，集群越动态，就越难使用集中式注册表准确跟踪服务的状态。

此外，所有的平衡 仍然 发生在客户端。这会导致负载均衡不协调的相同问题，这意味着服务上的实际负载可能会出现不均衡。此外，客户端负载平衡器 基于分布式服务中的指标 提供准确和诚实的请求协调系统是另一个挑战，这可能比前一个更难。 因此，我们必须找到更好的解决方案来构建反应式系统。

幸运的是，响应式 manifesto 为与服务器端和客户端平衡相关的问题提供了解决方案：

该声明可以解释为使用独立 消息代理来传输消息。考虑下图：

图 8.6。使用消息队列作为服务的负载平衡示例

在上图中，编号点的含义如下：

首先，所有请求都通过消息队列发送，然后可以将它们发送给可用的工作线程。此外，消息队列可以保持消息持续存在，直到其中一个工作人员请求新消息。通过这种方式，消息队列知道系统中有多少相关方，并可以根据该信息管理负载。反过来，每个工人都能够在内部管理背压并根据机器的可能性发送需求。仅通过监控待处理消息的数量，我们就可以增加活跃工作人员的数量。此外，仅仅依靠待处理的工人需求量，我们就可以减少休眠工人。

虽然消息队列解决了客户端负载均衡的问题，但似乎我们又回到了与之前非常相似的服务器端负载均衡解决方案，消息队列可能会成为系统中的热点.然而，事实并非如此。首先，这里的通信模型有点不同。消息队列不是搜索可用服务并决定将请求发送给谁，而是将传入消息放入队列中。然后，当工作人员声明接收消息的意图时，排队的消息将被传输。因此，这里有两个独立的、可能独立的阶段。这些如下：

另一方面，我们可以为每组收件人复制消息队列。因此，我们可以增强系统的可扩展性并巧妙地避免任何瓶颈。考虑下图：

图 8.7。消息队列即服务的弹性示例

编号 diagram 的每个部分解释如下：

在这里，我们为每个收件人组设置一个队列，并为该组中的每个队列设置一个复制集。但是，负载可能因组而异，因此一组可能会超载，而另一组可能只是处于休眠状态而没有任何工作，这可能是一种浪费。因此，我们可能会依赖支持虚拟队列的消息代理，而不是将消息队列作为单独的服务。通过这样做，我们可以降低基础设施的成本，因为系统上的负载可能会减少，因此一个消息代理可以在不同的接收者组之间共享。反过来，消息代理也可能是一个反应系统。因此，消息代理可以具有弹性和弹性，并且可以使用异步、非阻塞消息传递来共享其内部状态。考虑下图：

图 8.8。分布式消息代理的弹性

在上图中，编号点的含义如下：

上图描述了消息代理的可能设计，作为一个可以成为目标应用程序可靠主干的系统。从上图中可以看出，消息代理可以拥有系统需要的任意数量的虚拟队列。现代消息代理采用状态共享技术，例如最终一致性和消息多播，从而实现开箱即用的弹性。消息代理可以是异步消息传输的可靠层，具有背压支持和可重放性保证。

这对我们意味着什么？这意味着在消息代理崩溃的情况下，所有消息都可用，因此一旦消息传递层可用，所有未传递的消息都可以找到它们的目的地。

总之，我们可以得出结论，消息代理技术提高了系统的整体可扩展性。在这种情况下，我们可以很容易地构建一个弹性系统，因为消息代理可以作为一个反应系统。因此，沟通不再是瓶颈。

尽管使用 消息代理的想法可能是大多数业务需求的梦想，但我们的实际实现自己的消息代理可能会变成一场噩梦。幸运的是， 市场 现在 为我们提供了一些强大的开源解决方案。一些最流行的消息代理和消息平台包括RabbitMQ、Apache Kafka、Apache Pulsar、Apache RocketMQ、NATS、NSQ等。

2016年初，Spring Cloud引入了一个新的模块称为 Spring Cloud Data Flow，有 官方模块说明 在这个链接：https://cloud.spring.io/spring-cloud-dataflow:

概括地说，该模块的主要思想是实现功能业务转换的开发与已开发组件之间的实际交互之间的分离。换句话说，这是业务流程中功能及其组成之间的分离。为了解决这个问题，Spring Cloud Data Flow 为我们提供了一个用户友好的 Web 界面，这使得上传可部署的 Spring 成为可能启动应用程序。然后，它通过使用上传的工件并将组合管道部署到所选平台（例如 Cloud Foundry、Kubernetes、Apache YARN 或 Mesos）来设置数据流。此外，Spring Cloud Data Flow 为源（数据库、消息队列, 和文件），用于数据转换的不同内置处理器和接收器，它们表示存储结果的不同方式。

如前所述，Spring Cloud Data Flow 采用了流处理的思想。因此，所有部署的流都构建在 Spring Cloud Stream 模块之上，并且所有通信都是通过分布式、弹性消息代理（如 Kafka）或 RabbitMQ 的分布式高度可扩展变体完成的。

为了了解使用 Spring Cloud Data Flow 进行分布式响应式编程的强大功能，我们将构建一个支付处理流程。正如我们可能已经知道的那样，支付处理非常复杂。但是，这是此过程的简化图：

图 8.10。支付处理流程图

正如我们可能已经注意到的那样，用户的付款必须通过几个重要步骤，例如验证、帐户限制检查和付款批准。在 第 6 章中， WebFlux 异步非阻塞通信， 我们构建了一个类似的应用程序，其中一个服务编排了整个流程。尽管整个交互分布在对多个独立微服务的异步调用中，但流的状态由 Reactor 3 内部的一个服务持有。这意味着在该服务发生故障的情况下，恢复状态可能具有挑战性。

幸运的是，Spring Cloud Data Flow 依赖于 Spring Cloud Streams，后者依赖于弹性消息代理。因此，如果发生故障，消息代理将不会确认关于message 接收，因此 因此，消息将被重新传递给另一个执行者，而无需额外的努力。

由于我们已经对 Spring Cloud Data Flow 内部的核心原理和 Payment Flow 的业务需求有了基本的了解，我们可以使用该技术的堆栈来实现该服务。

首先，我们必须定义入口点，它通常可以作为 HTTP 端点访问。 Spring Cloud Data Flow 提供了一个可以使用 Spring Cloud Data Flow DSL 定义的 HTTP 源， 如下例所示：

SendPaymentEndpoint=Endpoint: http --path-pattern=/payment --port=8080

在前面的示例中，我们定义了一个新的数据流函数，它将所有 HTTP 请求表示为消息流。因此，我们可以以明确的方式对它们做出反应。

定义 HTTP 端点后，我们必须验证 incoming 消息。不幸的是，这部分流程包含实际的业务逻辑，并且需要对流程的某个阶段进行自定义实现。不过幸运的是，Spring Cloud Data Flow 允许在流程中使用自定义 Spring Cloud Stream applications。

一方面，我们可以为自己的独立 Spring Cloud Stream 应用程序提供自定义验证逻辑。但是，另一方面，我们仍然需要处理所有配置、uber-jars、较长的启动时间以及与应用程序部署相关的其余问题。幸运的是，使用 Spring Cloud Function 项目可以避免这些问题。

Spring Cloud Function 的主要目的是通过函数促进业务逻辑。该项目提供了将自定义业务逻辑与运行时细节分离的能力。因此，相同的功能可能会以不同的方式和地点重用。

在开始使用 Spring Cloud Function 之前，我们将在本节中介绍 Spring Cloud Function 模块的主要特性，并更好地了解其内部结构。

Spring Cloud Function 的核心是对可能运行在 Spring Cloud Streams、AWS Lambda 或使用任何通信传输的任何其他云平台之上的应用程序的额外抽象级别。

默认情况下，Spring Cloud Function 具有适配器，用于将函数可配置地部署到 AWS Lambda、Azure Functions 和 Apache OpenWhisk。与直接上传 Java 函数相比，使用 Spring Cloud Function 的主要好处是可以使用大多数 Spring 功能，并且不依赖于特定的云提供商 SDK。

Spring Cloud Function 提供的编程模型只不过是对以下 Java 类之一的定义——java.utils.function.Function、 java.utils.function.Supplier 和 java.utils.function.Consume。此外，Spring Cloud Function 可以在不同的框架组合中使用。例如，我们可以创建一个 Spring Boot 应用程序，它可以是一个平台元素的功能。反过来，它们中的一些可能 被表示为一个普通的Spring @Bean：

@SpringBootApplication                                             // (1)
@EnableBinding(Processor.class)                                    // (1.1)
public class Application {                                         //

   @Bean                                                           // (2)
   public Function<                                                //
      Flux<Payment>,                                               // 
      Flux<Payment>                                                //
   > validate() {                                                  //
      return flux -> flux.map(value -> { ... });                   // (2.1)
   }                                                               //

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

在上述代码中，numbered 点的含义如下：

从前面的例子可以看出，Spring Cloud Function 支持不同的编程模型。例如，支持 Reactor 3 反应类型和 Spring Cloud Streams 的消息转换，它将这些流暴露给外部目的地。

此外，Spring Cloud Function 不限于预定义的函数。例如，它有一个内置的运行时编译器，可以在属性文件中以字符串形式提供函数，如下例所示：

spring.cloud.function:                                             // (1)
   compile:                                                        // (2)
      payments:                                                    // (3)
         type: supplier                                            // (4)
         lambda: ()->Flux.just(new org.TestPayment())              // (5)

numbered 代码的每一部分解释如下：

前面的示例显示 Spring Cloud Function 提供了动态定义和运行函数的能力，而无需预先编译它们。这种能力可用于实现功能即服务（FaaS ) 软件解决方案中的功能。

除此之外，Spring Cloud Function 提供了一个名为 spring-cloud-function-task的模块，它允许使用相同的属性文件在管道中运行上述函数：

spring.cloud.function:
task:                                                           // (1)
supplier: payments                                           // (2)
function: validate|process                                   // (3)
consumer: print                                              // (4)
   compile:                                                        
      print:                                                    
type: consumer
         lambda: System.out::println
         inputType: Object
      process:                                                    
type: function
         lambda: (flux)->flux
         inputType: Flux<org.rpis5.chapters.chapter_08.scf.Payment>
         outputType: Flux<org.rpis5.chapters.chapter_08.scf.Payment>

代码中的编号解释如下：

正如我们所见，通过使用带有 Spring Cloud Function 模块作为依赖项的纯 Spring Boot 应用程序，可以运行用户准备好的函数并将它们组合在一个复杂的处理程序中。

 spring-cloud-function-compiler 模块在 Spring Cloud Function 生态系统中扮演着重要角色。除了从属性文件进行动态函数编译外，此模块还公开了 Web 端点，它 允许动态函数部署。 例如， 通过在终端中调用以下 curl 命令，我们可以将提供的函数添加到正在运行的 Spring Boot 应用程序中：

curl -X POST -H "Content-Type: text/plain" \
  -d "f->f.map(s->s.toUpperCase())" \
  localhost:8080/function/uppercase\
  ?inputType=Flux%3CString%3E\
  &outputTupe=Flux%3CString%3E

在此示例中，我们上传了一个函数，其中 Flux<String> 作为它的输入和输出类型。

通过 running 部署一个函数后，依赖于 spring-cloud 的轻量级 Spring Boot 应用程序-function-web and spring-cloud-function-compiler， 我们获得了基于 HTTP 的 on-the-fly 函数部署及其动态部署为一个单独的网络应用程序。例如，通过使用相同的 jar 文件并更改程序参数，我们可以使用不同的函数运行它，如下所示：

java -jar function-web-template.jar \
  --spring.cloud.function.imports.uppercase.type=function \
  --spring.cloud.function.imports.uppercase.location=\
file:///tmp/function-registry/functions/uppercase.fun \
\
  --spring.cloud.function.imports.worldadder.type=function \
  --spring.cloud.function.imports.worldadder.location=\
file:///tmp/function-registry/functions/worldadder.fun

在这个例子中，我们导入了两个函数：

从前面的示例代码可以看出，我们使用 spring.cloud.function.imports 命名空间来定义导入函数的名称（ 粗体），然后是它们的类型（斜体） 这些字节码的位置功能。应用程序启动成功后，我们可以通过执行以下 curl 命令来访问部署的函数：

curl -X POST -H "Content-Type: text/plain" \
   -d "Hello" \
   localhost:8080/uppercase%7Cworldadder

作为执行的结果 我们收到"HELLO World"，它确保这两个函数都存在于服务器上并按照URL中定义的顺序 执行。

同样，我们可以在相同或独立的应用程序中部署和导入其他函数。

或者，Spring Cloud Function 提供了一个部署模块，该模块扮演独立功能的容器的角色。在之前的案例中，我们能够运行内置函数或通过 spring-cloud-function-compiler web API 进行部署。我们已经了解了如何使用已部署的功能并将它们作为独立的应用程序运行。尽管具有这种灵活性，但 Spring Boot 应用程序的启动时间可能比该功能的执行时间长得多。在某些情况下（连同纯函数），我们必须使用一些 Spring Framework 库。例如，依赖 Spring Data 或 Spring Web features。因此，在这种情况下可能有用的是部署薄罐。 Spring Cloud Function 在这里提供了一个额外的模块，称为 spring-cloud-function-deployer。

Spring Cloud Function Deployer 模块允许运行 每个 jar 具有相同的 Spring Deployer 应用程序，但完全隔离。乍一看，使用该模块并没有获得宝贵的好处。然而，我们可能还记得，独立函数（这是我们想要实现的）在它们的引导和执行中是快速的。打包到 Spring Boot 环境中启动的函数需要启动整个 Spring Boot 应用程序，与函数的启动时间相比，这通常需要大量时间。

因此，为了解决这个问题，Spring Cloud Function Deployer 首先启动自己并预加载部分 JDK 类。然后它为每个带有函数的 jar 创建子 ClassLoader 。  每个 jar 的执行发生在它自己的 Thread 这使得并行执行成为可能。由于每个 jar 都是一个独立的微型 Spring Boot 应用程序，它在自己的 Spring 上下文中运行，因此 bean 不会与相邻应用程序的 bean 相交。最后，子 Spring Boot 应用程序的启动速度明显更快，因为父 ClassLoader 已经完成了 JVM 预热的艰巨工作。

此外， spring-cloud-function-deployer 和 spring-boot-thin-launcher的杀手锏组合 也可以解决 fat jar 问题。 Spring Boot Thin Launcher 是一个插件用于 Maven 和 Gradle，它 覆盖默认的 Spring Boot fat JarLauncher 并提供 ThinJarWrapper 和 < code class="literal">ThinJarLauncher 改为。这些类首先完成打包无依赖项 jar 所需的所有工作，然后——仅在引导阶段——它们从配置的缓存（例如，从本地 Maven 存储库）中找到所有需要的依赖项，或者从配置的 Maven 存储库。以这种方式运行，我们的应用程序可能会将 jar 的大小减少到几 KB，并将启动时间减少到数百毫秒。

总结 得到的关于Spring Cloud Function的信息 ，我们来看下面这张生态系统的概括图：

图 8.11。该  Spring Cloud Function 的生态系统

在上图中，numbered 点的含义如下：

正如我们所见，使用 Spring Cloud Function 模块以及与现有 Cloud 平台的直接集成，我们可以构建自己的 Function as a Service (FaaS) 平台，几乎提供了 Spring Framework 的全部功能，允许使用 lightweight 函数。但是，我们必须记住，当有实例部署、监控和管理的基础时，Spring Cloud Function 显示出强大的功能，因此可以在此基础上构建 Spring Cloud Function 生态系统并公开 FaaS 功能。因此，在下一节中，我们将介绍 Spring Cloud Function 如何与成熟的 Spring Cloud Data Flow 生态系统结合使用。

现在，对 Spring Cloud Function 生态系统有了足够的了解，我们可以回到主题，看看如何使用这个很棒的模块。有一个 additional 模块，名为 Spring Cloud Starter < /span>Stream App Function，可以在 Spring Cloud Data Flow 中使用 Spring Cloud Function 功能。这个模块允许我们使用纯 jars 并将它们部署为 Spring Cloud Data Flow 的一部分，而不需要 Spring 的任何 redundant 开销Boot. 由于我们这里有一个简单的映射，一个简化的 Validation 函数可以简化为一个传统的 Function 代码> 函数 可以如下所示：

public class PaymentValidator 
       implements Function<Payment, Payment> { 

  public Payment apply(Payment payment) { ... } 
}

打包发布一个工件后，我们或许可以编写如下流管道脚本，将我们的 HTTP 源连接到 Spring Cloud Function Bridge：

SendPaymentEndpoint=Endpoint: http --path-pattern=/payment --port=8080 | Validator: function --class-name=com.example.PaymentValidator --location=https://github.com/PacktPublishing/Hands-On-Reactive-Programming-in-Spring-5/tree/master/chapter-08/dataflow/payment/src/main/resources/payments-validation.jar?raw=true 

source.function=maven://org.springframework.cloud.stream.app:function-app-rabbit:1.0.0.BUILD-SNAPSHOT

source.function.metadata=maven://org.springframework.cloud.stream.app:function-app-rabbit:jar:metadata:1.0.0.BUILD-SNAPSHOT

processor.function=maven://org.springframework.cloud.stream.app:function-app-rabbit:1.0.0.BUILD-SNAPSHOT

processor.function.metadata=maven://org.springframework.cloud.stream.app:function-app-rabbit:jar:metadata:1.0.0.BUILD-SNAPSHOT

sink.function=maven://org.springframework.cloud.stream.app:function-app-rabbit:1.0.0.BUILD-SNAPSHOT

sink.function.metadata=maven://org.springframework.cloud.stream.app:function-app-rabbit:jar:metadata:1.0.0.BUILD-SNAPSHOT

最后，为了完成流程的第一部分，我们必须向不同的目的地提供经过验证的付款，并根据验证结果选择端点。由于验证函数是一个纯函数，不应该访问基础设施（例如 RabbitMQ 路由标头），我们应该将该责任委托给其他地方。幸运的是，Spring Cloud Data Flow 提供了 Router Sink，它允许基于如下表达式将传入消息路由到不同的队列：

...  | router --expression="payload.isValid() ? 'Accepted' : 'Rejected'"

或者，我们可以配置一个源来监听 particular 消息队列名称。例如，管道脚本负责监听名为 Accepted 的 RabbitMQ 通道，如下所示：

...
Accepted=Accepted: rabbit --queues=Accepted

根据付款流程图，付款处理的以下步骤将其状态保持在  接受 状态。通过这种方式，用户可以使用他们的付款访问特定页面，并检查每笔付款的处理状态。因此，我们应该提供与数据库的集成。例如，我们可以在 MongoDB 中存储付款转换的状态。 Spring Cloud Data Flow 提供了一个 MongoDB Sink。使用它，我们可以轻松地将传入消息写入 MongoDB。依靠 Spring Data Flow 插件，我们可以将消息广播到 MongoDB 接收器和下一个执行步骤。这种技术 只能 在完全可靠的消息代理（如Apache Kafka）的情况下是一个有效的解决方案。我们可能知道，卡夫卡 坚持 信息。因此，即使执行在某个阶段崩溃，消息也会在消息代理中可用。因此，MongoDB 拥有一个旨在在 UI 中使用的状态，而实际的处理状态则保存在  消息代理；因此，它可以在任何时间点重播。另一方面，对于快速、内存中的 message 代理，例如 RabbitMQ，依赖存储在 MongoDB 中的状态作为事实来源。因此，我们必须确保在执行下一步之前已经存储了付款的状态。不幸的是，为了实现这样的功能，我们必须编写一个自定义的 Spring Cloud Stream 应用程序，将 MongoDB 包装为流程中的一个处理阶段。

通过对流程的其余部分重复类似的操作，我们可以实现以下执行流程：

图 8.12。在 Spring Cloud Data Flow 用户界面的支持下完成了 Payment 的执行流程

前面的流可视化由以下管道脚本表示：

SendPaymentEndpoint=Endpoint: http --path-pattern=/payment --port=8080 | Validator: function --class-name=com.example.PaymentValidator --location=https://github.com/PacktPublishing/Hands-On-Reactive-Programming-in-Spring-5/tree/master/chapter-08/dataflow/payment/src/main/resources/payments.jar?raw=true | router --expression="payload.isValid() ? 'Accepted' : 'Rejected'"

Accepted=Accepted: rabbit --queues=Accepted | MarkAccepted: mongodb-processor --collection=payment | Limiter: function --class-name=com.example.PaymentLimiter --location=https://github.com/PacktPublishing/Hands-On-Reactive-Programming-in-Spring-5/tree/master/chapter-08/dataflow/payment/src/main/resources/payments.jar?raw=true | router --expression="payload.isLimitBreached() ? 'Rejected' : 'Checked'"

Checked=Checked: rabbit --queues=Checked | MarkChecked: mongodb-processor --collection=payment | Approver: function --class-name=com.example.PaymentApprover --location=https://github.com/PacktPublishing/Hands-On-Reactive-Programming-in-Spring-5/tree/master/chapter-08/dataflow/payment/src/main/resources/payments.jar?raw=true | router --expression="payload.isApproved() ? 'Approved' : 'Rejected'"

Approved=Approved: rabbit --queues=Approved | MarkApproved: mongodb-processor --collection=payment | Executor: function --class-name=com.example.PaymentExecutor --location=https://github.com/PacktPublishing/Hands-On-Reactive-Programming-in-Spring-5/tree/master/chapter-08/dataflow/payment/src/main/resources/payments.jar?raw=true | router --expression="payload.isExecuted() ? 'Executed' : 'Rejected'"

Executed=Executed: rabbit --queues=Executed | MarkExecuted: mongodb --collection=payment

Rejected=Rejected: rabbit --queues=Rejected | MarkRejected: mongodb --collection=payment

最后，通过部署该流，我们可以执行支付并在控制台中查看执行日志。

这里值得注意的一点是，deployment 流程就像业务逻辑开发一样简单。首先，Spring Cloud Data Flow 工具包构建在 Spring Cloud Deployer 之上，用于部署到 Cloud Foundry、Kubernetes、Apache Mesos 或 Apache YARN 等现代平台上。该工具包公开了允许配置应用程序源的 Java API（例如，Maven 存储库、artifactId、 groupId 和 version) 及其后续部署到目标平台。 除此之外， Spring Cloud Deployer 足够灵活，并提供更广泛的配置和属性列表，其中之一是可部署实例的副本数。

拥抱上述可能性，Spring Cloud Data Flow 提供一键单击（或一个终端命令）部署 能够传递所需的配置和属性。

总而言之，我们从 Spring Cloud Streams 的基础开始，以对几个模块的强大抽象结束了这个故事。结果，我们看到，在这些项目的支持下，我们可以通过应用反应式编程的不同抽象来构建反应式系统。 提到的 使用消息代理进行异步、可靠消息传递的技术涵盖了大多数业务需求。此外，该技术总体上可以降低反应式系统的开发成本，并可用于快速开发大型网络商店、物联网或聊天应用程序等系统。然而，重要的是要记住，尽管所描述的方法提高了系统可靠性、可伸缩性和吞吐量，但由于额外的通信，它可能会损害请求处理延迟，尤其是与持久消息代理的通信。因此，如果我们的系统可以容忍发送和接收消息之间的几毫秒的额外延迟，那么这种方法可能适用于我们的案例。

乍看之下，RSocket 似乎不是很创新；如今，我们已经拥有诸如 RxNetty 或 Reactor-Netty（Netty 的默认 WebFlux 包装器）之类的 Web 服务器。这些解决方案 提供了建立在反应流类型之上的API（允许从/向网络读取或写入），这意味着背压支持。然而，这种背压的唯一问题是它是孤立地工作的。随后，这意味着仅在组件和网络之间连接适当的背压支持。

例如，使用 Reactor-Netty，我们只有在准备好时才能使用传入的字节。同样，网络的准备情况 暴露为 Subscription#request 方法调用。这里的核心问题是组件的实际需求没有跨越网络边界，如下图所示：

图 8.13。孤立背压的例子

从上图我们可以看出，这里有两个服务 (Service A和 服务 B）。除此之外，我们还有一个抽象的 Adapter，它允许服务（业务逻辑）通过 Network. 在这个例子中，假设我们的适配器是 Reactor-Netty，它提供反应式访问。这意味着背压控制在传输级别正确实施。因此， 适配器会通知发布者  大约 此时它可以写入多少元素。在我们的例子中， Service A 创建一个continuous 与服务 B 的连接（例如，WebSocket 连接） > 一旦连接可用，它就会开始向适配器发送元素。因此，适配器会妥善处理向网络写入元素。在连接的另一端，我们有Service B，它通过同一个适配器使用来自网络的消息。作为Subscriber，Service B向适配器表达需求，并在适当数量的接收到字节后，适配器将它们转换为逻辑元素，并将其发送到业务逻辑。同样，一旦元素被发送到适配器，适配器会负责将其转换为字节，然后按照传输级别定义的流控制将它们发送到网络。

我们可能记得 第 3 章，Reactive Streams - 新 Streams 的标准，当消费者足够慢以至于生产者很容易溢出时，生产者和消费者之间的最坏情况就会出现消费者。为了强调隔离背压的问题，假设我们有一个慢的消费者和一个快的生产者，所以部分事件在Socket Buffer< /跨度>。不幸的是，缓冲区的大小是有限的，并且在某些时候，网络数据包可能会开始丢弃。当然，在许多方面，传输的行为取决于交互协议。例如，可靠的网络传输（如 TCP）具有包含 sliding 窗口和 确认。 这意味着，一般来说，背压是可以实现的在二进制级别（在接收到的字节和相应的确认级别）。在这种情况下，TCP 负责重新传递消息以及缺少确认会降低发布者速度的情况  Service A 。这里的缺点是对应用程序本身性能的影响很大，因为我们必须处理丢弃的包的重新交付。此外，通信的性能也可能会降低，因此系统的整体稳定性可能会丢失。尽管这样的传输流控制是可以接受的，但是连接的利用率很低。这是因为我们不能重用同一个连接来复用多个逻辑流。另一方面是当消费者无法遵循传输流控制时，消费者会继续在内部缓冲字节，这可能会以 OutOfMemoryError 告终。

与使用 Reactor Netty 或 RxNetty 可以轻松实现的隔离反应式 Publisher-Subscriber 通信相比，RSockets 提供了一个 二进制 protocol 是一种跨异步网络边界的反应流和语义的应用协议：

图 8.14。使用 RSocket 的背压示例

从上图中我们可以看出，通过 RSocket 协议，我们获得了对跨越网络边界传输需求的支持。因此，生产者服务可以响应请求并发送相应数量的onNext信号，这些信号也通过网络传输。此外，RSocket 工作在负责将数据写入网络的适配器之上，以便本地需求可以协调< /a> 与收到的来自消费者服务的需求隔离。一般来说，RSocket 是一种传输不可知论者，它与 TCP 一起支持 Aeron 和通过 WebSocket 进行的通信。

乍一看，似乎连接的使用效率很低，因为服务之间的交互可能很低。然而，RSocket 协议的强大功能之一是它允许为同一服务器和客户端之间的许多流重用相同的套接字连接。因此，它可以优化一个连接的使用。

与协议一起，RSockets 启用 对称交互模型，例如：

正如我们所见，RSocket 提供了大量常用的异步消息传递交互模型，一开始只获取单个连接。

RSocket 协议和 interaction 模型在 Java（以及 C++、JS、Python、和 Go）并在 Reactor 3 之上实现。RSocket-Java 模块提供了以下编程模型：

RSocketFactory                                                     // (1)
   .receive()                                                      // (1.1)
   .acceptor(new SocketAcceptorImpl())                             // (1.2)
   .transport(TcpServerTransport.create("localhost", 7000))        // (1.3)
   .start()                                                        // (1.4)
   .subscribe();                                                   //

RSocket socket = RSocketFactory                                    // (2)
   .connect()                                                      // (2.1)
   .transport(TcpClientTransport.create("localhost", 7000))        // 
   .start()                                                        //
   .block();                                                       // (2.2)

socket                                                             // (3)
   .requestChannel(                                                // (3.1)
      Flux.interval(Duration.ofMillis(1000))                       //
          .map(i -> DefaultPayload.create("Hello [" + i + "]"))    // (3.2)
   )                                                               //
   .map(Payload::getDataUtf8)                                      // (3.3)
   .doFinally(signalType -> socket.dispose())                      //
   .then()                                                         // 
   .block();                                                       //

编号代码的每个部分都在此处解释：

在前面的示例中，我们在客户端和服务器之间进行了有线双工通信。在这里，所有通信都是在对 Reactive Streams 规范和 Reactor 3 的支持下完成的。

此外，重要的是要提到 SocketAcceptor 的实现：

class SocketAcceptorImpl implements SocketAcceptor {               // (1)

@Override                                                       // (2)
public Mono<RSocket> accept(                                    // 
      ConnectionSetupPayload setupPayload,                         // (2.1)
RSocket reactiveSocket                                       // (2.2)
   ) {
return Mono.just(new AbstractRSocket() {                     // (3)
@Override                                                 // 
public Flux<Payload> requestChannel(                      // (3.1)
            Publisher<Payload> payloads                            // (3.2)
         ) {                                                       //
return Flux.from(payloads)                             // (3.3)
                       .map(Payload::getDataUtf8)                  //
                       .map(s -> "Echo: " + s)                     //
                       .map(DefaultPayload::create);               //
}                                                         //
      });                                                          //
}                                                               //
}

代码中的编号解释如下：

我们可以看到，SocketAcceptor的定义并不代表handler的定义。在这种情况下， SocketAcceptor#accept 方法的调用指的是一个新的传入连接。反过来，在 RSocket-Java 中，RSocket 接口同时表示客户端和服务器的处理程序。最后，各方之间的通信是点对点通信，这意味着双方都可以处理请求。

此外，为了实现可扩展性，RSocket-Java 提供了 RSocketLoadBalancer 模块，该模块可以与 Eureka 等服务注册中心集成。例如，以下 代码显示了与 Spring Cloud Discovery 的简单集成：

Flux                                                     
    .interval(Duration.ofMillis(100))                              // (1)
    .map(i ->
discoveryClient
          .getInstances(serviceId)                                 // (2)
          .stream()                                                //
          .map(si -> 
             new RSocketSupplier(() ->
RSocketFactory.connect()                           // (3)
                              .transport(                          //
                                 TcpClientTransport.create(        //
                                    si.getHost(),                  // (3.1)
                                    si.getPort()                   // (3.2)
                                 )                                 //
                              )                                    //
                              .start()                             //
             ) {
public boolean equals(Object obj) { ... }          // (4)
                                                                   //
public int hashCode() {                            //
return si.getUri().hashCode();                  // (4.1)
}                                                  //
             }
          )
          .collect(toCollection(ArrayList<RSocketSupplier>::new))
    )
    .as(LoadBalancedRSocketMono::create);                          // (5)

在前面的代码中，编号点的含义如下：

前面的例子展示了一个简单的方法 集成LoadBalancedRScoketMono 代码类="literal">DiscoveryClient。即使上面提到的例子效率不高，我们仍然可以学习如何正确使用 LoadBalancedRSocketMono。

概括地说，RSocket 是一种遵循 Reactive Streams 语义的通信协议，并通过背压控制支持扩展了跨网络边界的流式通信的新视野。反过来，有一个强大的基于 Reactor 3 的实现，它提供了一个简单的 API 用于在对等点之间建立连接，并在交互的生命周期中有效地利用它。

如果有一个名为 gRPC 的知名框架，我们可能仍然想知道为什么我们需要一个单独的框架。 gRPC 描述如下：

该项目最初由 Google 开发，旨在通过 HTTP/2 提供异步消息传递。它使用 Protocol Buffers (Protobuf) 作为 接口描述语言（IDL）和 作为其基础< /span> 消息 interchange 格式。

一般来说，gRPC 提供了与 Reactive Streams 几乎相同的 messaging 语义，并提供以下接口：

interface StreamObserver<V>  {

void onNext(V value);

void onError(Throwable t);

void onCompleted();
}

正如我们所见，语义是从 RxJava 1 到 Observer  的 1:1。反过来，gRPC 的 API 提供了 Stream< /code> 接口，扩展了以下方法： 

public interface Stream {

void request(int numMessages);

   ...

boolean isReady();

   ...
}

查看前面的代码，我们可能会感觉到 gRPC 以及异步消息传递也提供了背压控制支持。但是，这部分有点棘手。总的来说，交互流程有点类似于我们在 Diagram 8.13 中看到的，唯一的区别是它支持 flow control with更高的粒度。由于 gRPC 构建在 HTTP/2 之上，因此该框架使用 HTTP/2 流控制作为提供细粒度背压控制的构建块。尽管如此，流控制仍然依赖于以字节为单位的滑动窗口大小，因此逻辑元素级别粒度上的背压控制未被覆盖。

gRPC 和 RSocket 之间的另一个显着区别是 gRPC 是一个 RPC 框架，而 RSocket 是一个协议。 gRPC 基于 HTTP/2 协议，并为服务存根和客户端提供代码生成。默认情况下，gRPC 使用 Protobuf 进行消息传递格式；但是，它也可能支持 其他格式，例如JSON。同时，RSocket 只为服务端和客户端提供了一个响应式的实现。此外，还有一个名为 RSocket-RPC 的独立 RPC 框架，它建立在 RSocket 协议之上，并提供 gRPC 的所有功能。 RSocket-RPC 允许以与 gRPC 相同的方式基于 Protobuf 模型生成代码。因此，任何使用 gRPC 的项目都可以顺利迁移到 RSocket-RPC。

尽管该实现为使用 Reactor API 编写异步、低延迟和高吞吐量通信提供了更广泛 的可能性，但它留下了很多为开发人员使用基础设施配置。幸运的是，Spring 团队重视该项目，并开始尝试在 Spring 生态系统中采用如此出色的解决方案，并使用简化的编程模型而不是注解。

其中一项实验称为 Spring Cloud Sockets，旨在提供一种熟悉的（与 Spring Web 相比）编程模型来声明注释：

@SpringBootApplication                                             // (1)
@EnableReactiveSockets                                             // (1.1)
public static class TestApplication {                              //

   @RequestManyMapping(                                            // (2)
      value = "/stream1",                                          // (2.1)
      mimeType = "application/json"                                // (2.2)
   )                                                               //
   public Flux<String> stream1(@Payload String a) {                // (2.3)
      return Flux.just(a)                                          //
                 .mergeWith(                                       //
                    Flux.interval(Duration.ofMillis(100))          //
                        .map(i -> "1. Stream Message: [" + i + "]")//
                 );                                                //
   }                                                               //

   @RequestManyMapping(                                            // (2)
      value = "/stream2",                                          // (2.1)
      mimeType = "application/json"                                // (2.2)
   )                                                               //
   public Flux<String> stream2(@Payload String b) {                // (2.3)
      return Flux.just(b)                                          //
                 .mergeWith(                                       //
                    Flux.interval(Duration.ofMillis(500))          //
                        .map(i -> "2. Stream Message: [" + i + "]")//
                 );                                                //
   }                                                               //
}

代码中的编号在这里解释：

在这里，我们有熟悉的 Spring Boot 应用程序，它在 Spring Cloud Socket 的支持下，启用了 RSocket-Java 库的附加功能。反过来，Spring Cloud Sockets 也提供了 simplification 从客户端的角度与服务器交互：

public interface TestClient {                                      // (1)
   @RequestManyMapping(                                            // (2)
      value = "/stream1",                                          //
      mimeType = "application/json"                                //
   )                                                               //
   Flux<String> receiveStream1(String a);                          //

   @RequestManyMapping(                                            // (2)
      value = "/stream1",                                          //
      mimeType = "application/json"                                //
   )                                                               //
   Flux<String> receiveStream2(String b);                          //
}

在这里，我们所要做的就是在提供接口(1)时使用Spring Cloud Sockets声明一个RSocket Client。为了启用 RSocket 客户端，我们必须在   客户端的方法之上定义相同的注释，如在服务器示例中，并定义对应的处理程序路径。

因此，接口可以在运行时使用 ReactiveSocketClient工厂轻松转换为 代理，如下所示例子：

ReactiveSocketClient client = new ReactiveSocketClient(rSocket);
TestClient clientProxy = client.create(TestClient.class);

Flux.merge(
       clientProxy.receiveStream1("a"),
       clientProxy.receiveStream2("b")
    )
    .log()
    .subscribe();

在前面的示例中，我们创建了一个客户端（请注意，在此示例中，我们必须手动提供 RSocket 客户端的实例）。出于演示的目的，我们合并了两个 streams 并努力 .log()  结果。

如前所述，Spring Cloud Socket 模块是实验性的，原作者不再支持。尽管 Spring 团队继续进行内部实验并密切关注 RSocket，因为这是通过网络边界的 Reactive Streams 的强大解决方案，但还有一些其他框架也采用了 Java 协议的实现。

正如我们可能已经注意到贯穿这一节，RSocket 是构建高吞吐量和低延迟响应式的便捷方式基于异步对等通信的系统，采用反应式流规范。通常，RSocket 协议旨在减少感知延迟并提高系统效率。这可以通过双工连接支持非阻塞通信来实现。除此之外，RSocket 的设计目的是减少硬件占用空间。 反过来，RSocket 是一种可以用任何语言实现的协议。 最后，RSocket 在 Java 中的实现是建立在 Project Reactor 之上的，它提供了一个开箱即用的强大编程模型。

总的来说，RSocket 社区正在飞速发展，看起来很有希望。目前，该项目由 Facebook 和 Netifi 积极维护，在不久的将来，其他公司也将加入。