Spring Webflux的响应式Java简介

在本文中，您将使用Spring Boot和Spring Webflux创建一个反应式web服务。web服务将演示如何使用Split的javasdk在反应式环境中使用特性标志。您将使用Spring initializer项目快速地用必要的配置参数引导应用程序。您将构建的简单应用程序将公开一个资源，该资源每秒从James Baldwin的《致我侄子的一封信》中的一段话流式传输一次单词。

Java中的反应式编程

反应式编程是一种围绕流处理构建的编程范式。数据作为通过函数或运算符管道的数据点流进行处理。每个操作符都可以在将数据传递给下一个操作符之前对其进行转换和操作。您可能熟悉map、collect和reduce等操作符。这些操作被链接起来，直到转换后的数据被返回。反应式代码与命令式代码相反，命令式代码维护程序状态，函数或方法对状态进行操作，对状态进行变异。

反应式编程也是非阻塞和异步的。这意味着它被设计用于单线程、非阻塞服务器，如Netty、Undertow和Node.js。非阻塞代码是另一个巨大的主题，但简单地说，它是从不停止并等待某些事情完成的代码，例如从磁盘读取或进行网络调用，“阻塞”线程。相反，回调和承诺用于延迟代码执行，直到输入数据就绪，从而允许线程继续处理其他工作。

理想情况下，这允许实现超高性能，因为服务器可以在单个线程上完成所有工作，并且不会浪费时间和资源来管理处理器必须不断切换的工作线程的循环堆栈。然而，与所有事情一样，也有取舍。

例如，非阻塞代码在历史上一直难以实现、维护和调试（在互联网上搜索“回调地狱”）。随着技术的成熟，这在一定程度上发生了变化。然而，这个世界很少是理想的。不是所有的非阻塞代码都是非阻塞的，依赖项中一段写得不好的代码可能会阻塞整个web服务，从而影响性能。但是如果做得好，反应式非阻塞代码可以是一个很好的工具。非阻塞、反应式代码优于流处理的一个用例。

Spring Webflux是Spring Boot的反应式、非阻塞的web应用程序实现。它使用project reactor实现其反应式Java实现，默认情况下，Netty实现其非阻塞web服务器。Webflux实际上支持其他web服务器实现，例如Tomcat、Jetty、Undertow和Servlet3.1+容器。为什么有人会在Webflux中使用Tomcat或Jetty（使用带有反应式异步框架的阻塞servlet容器）是一个很好的开放性问题，但您有这个选择。

Spring Webflux隐式地与spring mvc形成对比。spring mvc是一种优秀的、老式的基于servlet的Java服务器。它是一个多线程并发系统，允许阻塞、同步代码。它不使用回调和承诺来处理并发性，而是假设每个请求将在自己的线程中发生（并且可能根据需要启动额外的工作线程）。事实上，这种范式或技术绝对没有错。对于每一篇声称非阻塞比阻塞快的基准文章，您可以找到另一篇声称相反的文章。一般来说，编写良好的阻塞代码可以和非阻塞代码一样快。一个设计良好、管理得当的系统将击败一个考虑不周、没有得到正确维护或管理的系统。

Java中的功能标志入门

深入研究代码之前的最后一个主题是特性标志。在传统的开发环境中，如果要对已部署的应用程序进行更新或更改，则需要更改代码、打包、部署并发布。这太慢了。特性标志是使发布过程更加动态的一种方法。本质上，它们是在运行时检查的动态变量，用于确定代码在部署时的行为。标志的状态存在于服务器上，并且状态可以实时更新。您可以在逻辑块中编写代码，逻辑块的状态由标志决定。

例如，使用这个系统，您可以将一个新特性作为alpha测试发布到一小部分用户群中。一旦通过了测试，你就可以把它作为一个测试版推广到更大的范围。如果测试失败，您可以回滚所有用户。如果通过，您可以完成向所有用户的推出。

另一个想法是将它们用于A/B测试。也就是说，动态地划分用户并将一组特性推出给一组用户，将另一组特性推出给不同的组。

在本教程中，您将使用Split作为功能标志服务提供程序。Split提供了一个免费帐户，允许您创建特性标志，并使用Split的javasdk将它们集成到Spring Boot应用程序中。

依赖项

Java：我在本教程中使用了java12。您可以通过adapteopenjdk网站下载并安装Java。或者你可以使用一个版本管理器，比如SDKMAN，甚至是自制的。

Split：注册一个免费的分割帐户，如果你还没有一个。这就是实现特性标志的方法。

HTTPie：这是一个强大的命令行HTTP请求实用程序，您将使用它来测试被动服务器。根据他们网站上的文档安装。

使用Spring初始化器创建演示应用程序

Spring有一个名为Spring initializer的项目可以帮助创建应用程序。您将使用它下载一个预配置的启动程序项目。

打开此链接：https://start.spring.io/，将加载正确的配置。

出于好奇，您可以通过命令行完成与上面链接相同的操作：

http https://start.spring.io/starter.tgz \
type==gradle-project \
language==java \
platformVersion==2.4.0.RELEASE \
packaging==jar \
javaVersion==11 \
groupId==com.example \
artifactId==demo \
name==demo \
description==Demo%20project%20for%20Spring%20Boot \
packageName==com.example.demo \
dependencies==devtools,webflux \
| tar -xzvf -

最重要的是配置webflux依赖项。如果您想了解有关配置选项的更多信息，可以查看Spring initializer GitHub页面。

我要指出的一个很酷的特性是页面底部的EXPLORE按钮。点击它，你就可以在网上浏览这个项目的预览了。

单击GENERATE按钮。下载demo.zip文件并将其解压缩到适当的位置。

创建反应式流服务器

在这一点上，您有一个功能齐全的反应式Spring引导应用程序，它完全不起任何作用。如果你在这一点上检查这个项目，你会发现它没有什么大不了的。基本上有一个DemoApplication.java文件。

@SpringBootApplication
public class DemoApplication {

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

}

当然，这种简单性在后端隐藏了大量的复杂性。Spring为您配置了一个应用程序，它使用Netty（这是默认的，还有其他选项，比如Undertow）作为web服务器引擎，并支持Java Reactor框架进行反应式Java编程。所有这些都已被折叠到Spring编程框架中，以实现高度可配置和可扩展性。

为了让生活更精彩，您可能应该让演示应用程序做些什么。在这种情况下，你要让它直播一个演讲，每秒钟发出一个新词。

创建一个名为DemoResource.Java的新Java文件。

package com.example.demo;

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;

@RestController
public class DemoResource {

    // Excerpt of James Baldwin "A Letter to my Nephew"
    private String[] speech = {
            "Well,","you","were","born;","here","you","came,","something","like","fifteen","years","ago,","and","though",
            "your","father","and","mother","and","grandmother,","looking","about","the","streets","through","which","they",
            "were","carrying","you,","staring","at","the","walls","into","which","they","brought","you,","had","every",
            "reason","to","be","heavy-hearted,","yet","they","were","not,","for","here","you","were,","big","James,",
            "named","for","me.","You","were","a","big","baby.","I","was","not.","Here","you","were","to","be","loved.",
            "To","be","loved,","baby,","hard","at","once","and","forever","to","strengthen","you","against","the",
            "loveless","world.","Remember","that.","I","know","how","black","it","looks","today","for","you.","It","looked",
            "black","that","day","too.","Yes,","we","were","trembling.","We","have","not","stopped","trembling","yet,",
            "but","if","we","had","not","loved","each","other,","none","of","us","would","have","survived,","and","now",
            "you","must","survive","because","we","love","you","and","for","the","sake","of","your","children","and",
            "your","children's","children.",
    };

    @GetMapping(value="/speech", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> getSpeech() {
        return Flux
                .fromArray(speech)
                .delayElements(Duration.ofSeconds(1))
                .repeat()
                .log();
    }
}

从shell启动Spring Boot应用程序。

./gradlew bootRun

应用程序完成从第二个shell加载后，使用HTTPie测试端点。注意--stream参数。很高兴HTTPie支持流式处理端点。

http :8080/speech --stream
Code language: SQL (Structured Query Language) (sql)

你应该看到下面这样的东西，詹姆斯·鲍德温的“给我侄子的信”中的一个词每秒都在流。

HTTP/1.1 200 OK
Content-Type: text/event-stream;charset=UTF-8
transfer-encoding: chunked

data:Well,

data:you

data:were

创建特征标记处理

在更改代码以使用javasplitsdk对治疗做出反应之前，需要在Split帐户中创建一个treatments。如果拆分和treatments对你来说是新的，我建议你通读一下Split网站上的入门信息。

简单地说，split是代码中的一个决策点，可以用来实时修改代码，而无需重建和重新部署代码。它是您在代码中定义的一个标志，允许您通过拆分仪表板更改代码的工作方式，而不是将更新推送到应用程序。处理是这些实时标志的特定状态。当您调用getTreatment（String key，String splitName）方法时，SplitClient根据键和split的名称检索处理。密钥可以是任意的、唯一的字符串。但通常可能是当前用户的用户名。

专业提示：有一个可选的第三个getTreatment参数（我们在这里不使用），它是一个属性映射对象，在名称-值对中包含用户属性。即使是敏感的用户数据也可以在这个映射中传递，因为没有一个数据被发送到Split的云。相反，属性映射在本地内存中与您在分割UI中输入的目标规则进行比较。更多关于splitsdk文档：使用属性映射定制目标。

例如，使用Split和feature标志，您可以在不同的用户子集上动态测试一组新功能，并且根据测试的进行方式，您可以将这些功能移动到完全展开或回滚，而无需重新部署应用程序。

我掩盖了上面的一个简化。我说过SplitClient.getTreatment（）方法检索处理。创建客户端时，缓存可能的处理值，并打开连接到Split的套接字。如果在拆分服务器上更新了处理，则使用套接字更新缓存的值。通过这种方式，getTreatment（）方法的性能要比它必须向拆分的服务器发出实际的网络请求时好得多，并且可以实时推送处理的更新（理论上是实时的，实际上我注意到仪表板上的更新和本地反映的更新之间大约有10-20秒的延迟）。

您将创建一个名为speech的拆分。split的处理将有三个可能的值：a、b和off。用户名将决定返回哪个处理（为简单起见，作为查询参数传入资源服务器）。

打开拆分仪表板。您应该在默认工作区中。

单击蓝色的创建拆分按钮。

单击“创建”。

您应该处于新拆分的暂存默认环境中。

单击添加规则按钮。默认情况下，“拆分”添加“开”和“关”。单击“添加治疗”按钮添加新值。将三个治疗值的标签更改为a、b和off。

现在需要定义两个目标规则。第一条规则将告诉Split，如果用户的名字是bob或fred，那么它应该作为一种治疗。

向下滚动，直到看到“设置目标规则”部分。单击添加规则。单击“在段中”下拉菜单，将光标悬停在字符串上，然后单击“在列表中”。在右侧的文本框中，输入两个名称：bob和fred。确保发球下拉列表（规则的灰色框右下角）显示a处理。

第二条规则将告诉Split，如果用户的名字是wilma或ted，它应该服务于治疗b。

单击添加规则。单击“在段中”下拉菜单，将光标悬停在字符串上，然后单击“在列表中”。在右侧的文本框中，输入两个名称：威尔玛和特德。确保serve下拉列表显示b处理。

在下面，您希望使任何其他用户的默认行为处于关闭状态。在下一节中，设置默认规则，将“发球”下拉列表更改为“关闭”。同样，在下一节中，设置默认处理方式，选择关闭处理方式。

单击屏幕顶部的蓝色保存更改按钮。向下滚动并单击确认。

就在拆分仪表板上。总而言之，您创建了一个包含三种处理方式的单一拆分。您创建了两个规则，将治疗a分配给名为bob或fred的用户，将治疗b分配给名为wilma或ted的用户。所有其他用户都将获得off治疗，如果存在网络问题或客户端由于其他原因无法检索治疗，这也是默认设置。

配置反应式流服务器以使用功能标记处理

打开build.gradle文件并添加Split-SDK依赖项。

dependencies {
...
    compile 'io.split.client:java-client:4.1.0'
}

创建一个SplitConfig文件来处理splitclientbean的配置，它将被注入DemoResource。

@Configuration
public class SplitConfig {

    @Value("#{ @environment['split.api-key'] }")
    private String splitApiKey;

    @Bean
    public SplitClient splitClient() throws Exception {
        SplitClientConfig config = SplitClientConfig.builder()
                .setBlockUntilReadyTimeout(1000)
                .enableDebug()
                .build();

        SplitFactory splitFactory = SplitFactoryBuilder.build(splitApiKey, config);
        SplitClient client = splitFactory.client();
        client.blockUntilReady();

        return client;
    }

}

您需要拆分API密钥。从Split dashboard，转到dashboard左上角的square workspace图标（可能默认为DE）并单击它。点击管理设置。单击左侧面板中“工作区设置”下的API键。

将有四个API键，两个用于生产，两个用于暂存。服务器端和客户端使用的密钥也不同。复制SDK和暂存默认密钥。

将其添加到application.properties。

split.api-key={yourSplitApiKey}

最后，更新DemoResource.java以使用SplitClient获取并使用它。

@RestController
public class DemoResource {

    // Excerpt of James Baldwin "A Letter to my Nephew"
    private String[] speech = {
            "Well,","you","were","born;","here","you","came,","something","like","fifteen","years","ago,","and","though",
            "your","father","and","mother","and","grandmother,","looking","about","the","streets","through","which","they",
            "were","carrying","you,","staring","at","the","walls","into","which","they","brought","you,","had","every",
            "reason","to","be","heavy-hearted,","yet","they","were","not,","for","here","you","were,","big","James,",
            "named","for","me.","You","were","a","big","baby.","I","was","not.","Here","you","were","to","be","loved.",
            "To","be","loved,","baby,","hard","at","once","and","forever","to","strengthen","you","against","the",
            "loveless","world.","Remember","that.","I","know","how","black","it","looks","today","for","you.","It","looked",
            "black","that","day","too.","Yes,","we","were","trembling.","We","have","not","stopped","trembling","yet,",
            "but","if","we","had","not","loved","each","other,","none","of","us","would","have","survived,","and","now",
            "you","must","survive","because","we","love","you","and","for","the","sake","of","your","children","and",
            "your","children's","children.",
    };

    SplitClient splitClient;

    public DemoResource(SplitClient splitClient) {
        this.splitClient = splitClient;
    }

    @GetMapping(value="/speech", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<List<String>> getSpeech(@RequestParam String user) {
        return Flux
                .fromArray(speech)
                .publishOn(Schedulers.boundedElastic())
                .map(word -> {
                    String treatment = splitClient.getTreatment(user,"speech");
                    String treatedWord = word;
                    if (treatment.equals("a")) {
                        treatedWord = word.toUpperCase();
                    }
                    else if (treatment.equals("b")) {
                        treatedWord = word.toLowerCase();
                    }
                    return Arrays.asList(treatedWord, treatment);

                })
                .delayElements(Duration.ofSeconds(1))
                .repeat()
                .log();
    }
}

注意，getSpeech（）方法现在返回字符串列表。它返回当前单词和处理。这只是为了测试或演示的目的，以便您可以明确地看到什么治疗是积极的。

另外，请注意，您正在使用Spring的依赖注入将SplitConfig类中配置的splitclientbean传递给DemoResource构造函数。

SplitClient splitClient;

public DemoResource(SplitClient splitClient) {
    this.splitClient = splitClient;
}

请注意getSpeech（）方法代码中的下面一行。这很重要。

.publishOn(Schedulers.boundedElastic())

必须使用publishOn运算符，因为SplitClient.getTreatment（）方法是阻塞或同步的。publishOn操作符将执行线程推送到一个新线程，这样它就不会阻塞Netty服务器使用的一个主线程。

请记住，非阻塞和异步服务器使用单个主线程（或非常少的主线程）工作。阻塞该线程会阻塞整个web服务器。永远不要这样做。如果你挡住了主线，你的同事会恨你，你的敌人会嘲笑你，你的朋友会嘲笑你。像Netty和Node.js这样的非阻塞服务器可能实现的高性能很大程度上是因为它们不必在线程之间执行大量上下文切换。但是，这取决于主线程中的代码的行为，即从不阻塞线程，而是使用回调和承诺延迟执行。

当您想在非阻塞上下文中使用阻塞API时，处理它的唯一方法就是将其推送到工作线程上。这里有一个权衡，因为一旦开始将工作从主线程推到其他线程上，就开始降低非阻塞模型的效率。你做的越多，你就越像一个传统的web服务器。在某种程度上，只需使用Tomcat（在springmvc中默认使用）这样的服务器，您可能会得到更好的服务，它经过了高度优化，可以处理多线程web环境。

最后，看一下getSpeech（）方法的一部分，该部分已添加用于检索和使用处理。

.map(word -> {
    String treatment = splitClient.getTreatment(user, "speech");
    String treatedWord = word;
    if (treatment.equals("a")) {
        treatedWord = word.toUpperCase();
    }
    else if (treatment.equals("b")) {
        treatedWord = word.toLowerCase();
    }
    return Arrays.asList(treatedWord, treatment);

})

您将看到对SplitClient的getTreatment（）调用。这是检索治疗值的阻塞调用。处理将是一个字符串值，可以是a、b或off。

之后，有一个非常简单的if/elseif序列，它根据处理修改返回的单词，并返回处理本身。

试试看。从shell运行服务器。

./gradlew bootRun

准备好之后，发出流式GET请求。

http :8080/speech user==bob --stream

您将看到一个包含所有大写字母的流，处理值为a。

HTTP/1.1 200 OK
Content-Type: text/event-stream;charset=UTF-8
transfer-encoding: chunked

data:["WELL,","a"]

data:["YOU","a"]

data:["WERE","a"]

data:["BORN;","a"]

保持Java服务器运行，但control-c停止HTTPie客户机。重新运行客户端，为用户名传入一个新值。

http :8080/speech user==wilma --stream
Code language: SQL (Structured Query Language) (sql)

这一次你会看到 treatment b和所有小写。

HTTP/1.1 200 OK
Content-Type: text/event-stream;charset=UTF-8
transfer-encoding: chunked

data:["well,","b"]

data:["you","b"]

data:["were","b"]

data:["born;","b"]

要查看如何实时操作这些治疗方法，请保持此运行并返回到您的拆分仪表板。打开语音分割并向下滚动到设置目标规则面板。在第二条规则（威尔玛和泰德的规则）中，将发球下拉列表从b切换到a。在这一点上，一切都没有改变。您必须单击屏幕顶部的保存更改按钮。确认更改。

现在查看shell窗口中的流值。几秒钟后，它将更改以反映此更新。

data:["at","b"]

data:["once","b"]

data:["and","b"]

data:["FOREVER","a"]

data:["TO","a"]

data:["STRENGTHEN","a"]

很酷。实时更新。显然，这些示例有些琐碎，但是一旦您了解了代码中嵌入的动态特性标志的威力，就可以用它们做各种各样的事情。

在本文中，您使用Spring初始化器创建了一个使用springboot和springwebflux的反应式Java应用程序。Java应用程序使用反应式、非阻塞代码生成一个词流。您使用HTTPie从命令行测试这个反应式流服务器。您使用Split在代码中实现了一个功能标志。您了解了如何创建拆分、定义处理方法以及基于用户属性定义目标规则。接下来，您使用splitjavasdk将这个特性标记集成到反应式javaweb应用程序中。您看到了如何使用publishOn操作符将阻塞代码正确地集成到非阻塞web环境中。最后，您看到了如何在Split仪表板上实时更新治疗值和目标规则。

您可以从GitHub repo上的这篇文章中找到示例的完整源代码：https://github.com/splitio-examples/split-webflux-example

原文地址：https://www.split.io/blog/reactive-java-spring-webflux/