什么是Reactive Streams in Java 译

函数式编程对于Reactive Programming很重要，但我不会在这篇文章中深刻探讨函数式编程。

在这篇文章中，我想看看Java中的总体Reactive发展环境。

Reactive Programming vs. Reactive Streams

有了这些新的流行语，就很容易对它们的含义感到困惑。

反应式编程是一种编程范式，但我不会称之为新的。它实际上已经存在了一段时间。

就像面向对象编程，函数式编程或过程式编程同样，反应式编程只是另外一种编程范式。

另外一方面，Reactive Streams是一种规范。对于Java程序员，Reactive Streams是一个API。Reactive Streams为咱们提供了Java中的Reactive Programming的通用API。

Reactive Streams API是Kaazing，Netflix，Pivotal，Red Hat，Twitter，Typesafe等众多工程师之间合做的产物。

Reactive Streams很是相似于JPA或JDBC。二者都是API规范。您须要使用API​​规范的实现。

例如，从JDBC规范中，您具备Java DataSource接口。Oracle JDBC实现将为您提供DataSource接口的实现。正如Microsoft的SQL Server JDBC实现也将提供DataSource接口的实现。

如今，您的高级程序能够接受DataSource对象，而且应该可以使用数据源，而没必要担忧它是由Oracle提供仍是由Microsoft提供。

就像JPA或JDBC同样，Reactive Streams为咱们提供了一个咱们能够编写代码的API接口，而无需担忧底层实现。

Reactive Programming

关于Reactive programming是什么，有不少意见。Reactive programming也有不少炒做！

开始学习Reactive programming范例的最佳起点是阅读“Reactive Manifesto”。

Reactive Manifesto

Reactive Manifesto描述了reactive systems的四个关键属性：

Responsive

若是可能的话，系统及时响应。响应性是可用性和实用性的基石，但更重要的是，响应性意味着能够快速检测到问题并有效地处理问题。reactive systems专一于提供快速一致的响应时间，创建可靠的上限，以便提供一致的服务质量。这种一致的行为反过来简化了错误处理，创建了最终用户的信心，并鼓励进一步的交互。

Resilient

系统在出现故障时保持响应。这不只适用于高可用性，关键任务系统 - 任何无弹性的系统在发生故障后都不会响应。经过复制，遏制，隔离和委派来实现弹性。故障包含在每一个组件中，将组件彼此隔离，从而确保系统的各个部分能够在不损害整个系统的状况下发生故障和恢复。将每一个组件的恢复委派给另外一个（外部）组件，并在必要时经过复制确保高可用性。组件的客户端不会负担处理其故障的负担。

Elastic

系统在不一样的工做负载下保持响应。反应系统能够经过增长或减小分配用于服务这些输入的资源来对输入速率的变化做出反应。这意味着设计没有争用点或中心瓶颈，从而可以分片或复制组件并在它们之间分配输入。Reactive Systems经过提供相关的实时性能测量来支持预测和反应式扩展算法。它们在商用硬件和软件平台上以经济高效的方式实现弹性。

Message Driven

Reactive Systems依靠异步消息传递在组件之间创建边界，以确保松散耦合，隔离和位置透明性。此边界还提供将故障委派为消息的方法。使用显式消息传递能够经过整形和监视系统中的消息队列并在必要时应用反压来实现负载管理，弹性和流量控制。做为通讯手段的位置透明消息传递使得管理失败能够在群集内或单个主机内使用相同的构造和语义。非阻塞通讯容许接收者仅在活动时消耗资源，从而减小系统开销。

前三个属性（Responsive, Resilient, Elastic）与您的架构选择更相关。很容易理解为何微服务，Docker和Kubernetes等技术是Reactive系统的重要方面。在单个服务器上运行LAMP堆栈显然不符合Reactive Manifesto的目标。

Message Driven and Reactive Programming

做为Java开发人员，它是咱们最感兴趣的最后一个属性，即Message Driven属性。

消息驱动的架构固然不是革命性的。若是你须要一个关于消息驱动系统的入门读物，我想建议阅读Enterprise Integration Patterns - 一本真正标志性的计算机科学书籍。本书中的概念为Spring Integration和Apache Camel奠基了基础。

咱们对Java开发人员感兴趣的Reactive Manifesto的几个方面是：消息失败，背压和非阻塞。这些是Java中反应式编程的微妙但重要的方面。

Failures as Messages

一般在Reactive编程中，您将处理消息流。

不但愿的是抛出异常并结束消息流的处理。首选方法是优雅地处理故障。

也许你须要执行一个Web服务而且它已经关闭了。也许你可使用备份服务？或者可能在10ms内重试？

我不会在这里解决每个边缘案例。关键的一点是，您不但愿因运行时异常而大声失败。理想状况下，您须要记录失败，并具备某种类型的重试或恢复逻辑。一般，故障是经过回调来处理的。

JavaScript开发人员习惯于使用回调。但回调可能会变得难以使用。JavaScript开发人员将此称为回调地狱。

在Reactive Steams中，异常是一等公民。不是粗暴抛出的。错误处理内置于Reactive Streams API规范中。

Back Pressure

你据说过“从消防中喝水”这句话吗？

背压是反应式编程中很是重要的概念。它为下游客户提供了一种说法，“我还有更多，请。”

想象一下，若是您正在查询数据库，结果集将返回1000万行。传统上，数据库将以客户端接受它们的速度吐出全部1000万行。

当客户端不能再接受时，它会阻塞。数据库焦急地等待着。阻止。链中的线程耐心地等待解除阻塞。

在Reactive世界中，咱们但愿咱们的客户可以说给我前1000个。而后咱们能够给他们1000并继续咱们的业务 - 直到客户回来并要求另外一组记录。

这与客户没有发言权的传统系统造成鲜明对比。经过阻塞线程来完成限制，而不是以编程方式。

Non-Blocking

对Java开发人员来讲重要的Reactive架构的最后，也许是最重要的方面是非阻塞的。

直到Reactive来了很长时间，没有阻塞彷佛并无那么重要。

做为Java开发人员，咱们已经被教会经过使用线程来利用强大的现代硬件。愈来愈多的核心意味着咱们可使用愈来愈多的线程。所以，若是咱们须要等待数据库或Web服务返回，则不一样的线程能够利用CPU。这彷佛对咱们有意义。当咱们被阻塞的线程等待某种类型的I / O时，不一样的线程可使用CPU。

所以，阻塞并非什么大问题。对？

系统中的每一个线程都将消耗资源。每次线程被阻塞时，都会消耗资源。虽然CPU在维护不一样线程方面很是有效，但仍然存在成本问题。咱们Java开发人员多是一群傲慢的人。

他们老是看不起JavaScript，有点讨厌的小语言。事实上，JavaScript共享“Java”这个词老是让咱们Java程序员感受有点脏。若是您是Java开发人员，当您必须指出Java和JavaScript是两种不一样的语言时，您有多少次感到烦恼？而后Node.js出现了。Node.js在吞吐量方面提出了疯狂的基准测试。而后Java社区注意到了。是的，脚本小子已经长大，正在侵占咱们的地盘。并非在谷歌的V8 JavaScript引擎中运行的JavaScript对于编程来讲是一个很是快速的天赐之物。Java曾经在性能方面有其瑕疵，但与现代本地语言相比，它很是高效。

Node.js表现的秘诀是无阻塞。

Node.js使用具备有限数量线程的事件循环。虽然在Java世界中的阻塞一般被视为没什么大不了的，但在Node.js世界中，它将成为性能的死亡之吻。

这些图形能够帮助您可视化差别。在Node.js中，有一个非阻塞事件循环。请求以非阻塞方式处理。线程不会卡在等待其余进程。

将Node.js模型与Java中使用的典型多线程服务器进行对比。并发是经过使用多个线程实现的。因为多核处理器的增加，这被广泛接受。

我我的认为这两种方法之间的区别在于高速公路和许多带灯光的城市街道之间的区别。

经过单线程事件循环，您的过程能够在超级高速公路上快速巡航。在多线程服务器中，您的进程会停在城市街道上并中止流量。二者均可以带来大量流量。可是，我宁愿在高速公路上巡航！

当您转向非阻塞范例时，您的代码会在CPU上停留更长时间。线程切换较少。您不只要管理许多线程，还要删除线程之间的上下文切换。

您将在系统容量中看到更多的空间，供您的程序使用。非阻塞不是圣杯，你不会看到事情变得更快。是的，管理阻塞须要付出代价。但考虑到全部事情，它相对有效。事实上，在一个适度使用的系统上，我不肯定差别会有多大可衡量。

可是，随着系统负载的增长，您能够期待看到的是，您将拥有额外的容量来为更多请求提供服务。您将得到更高的并发性。多少？好问题。用例很是具体。与全部基准同样，您的里程也会有所不一样。

The Reactive Streams API

咱们来看看Reactive Streams API for Java。Reactive Streams API仅包含四个接口。

Publisher

发布者是可能无限数量的有序元素的提供者，根据从其订阅者收到的需求发布它们。

public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}

Subscriber

在将订阅者实例传递给Publisher.subscribe（订阅者）以后，将接收对Subscriber.onSubscribe（订阅）的调用。
public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

Subscription

订阅表示订阅发布者的订阅者的一对一辈子命周期。

public interface Subscription {
    public void request(long n);
    public void cancel();
}

Processor

Processor表明一个处理阶段 - 既是订阅者又是发布者，并遵照二者的合同。

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

Reactive Streams Implementations for Java

Java中的Reactive环境正在不断发展和成熟。David Karnok在Advanced Reactive Java上有一篇很棒的博客文章Advanced Reactive Java，他将各类反应项目分解成几代。我会注意到下面每一代（随着新版本的发布，它可能会随时改变）。

RxJava

RxJava是ReactiveX项目中的Java实现。在撰写本文时，ReactiveX项目实现了Java，JavaScript，.NET（C＃），Scala，Clojure，C ++，Ruby，Python，PHP，Swift等等。

ReactiveX在GoF Observer模式上提供了反应性扭曲，这是一种很好的方法。ReactiveX调用他们的方法'Observer Pattern Done Right'

RxJava早于Reactive Streams规范。虽然RxJava 2.0+确实实现了Reactive Streams API规范，但您会注意到术语略有不一样。David Karnok是RxJava的关键提交者，他认为RxJava是第三代reactive library。

Reactor

Reactor是Pivotal的Reactive Streams兼容实现。

从Reactor 3.0开始，Java 8或更高版本是必需的。Spring Framework 5中的Reactive功能基于Reactor 3.0构建。Reactor是reactive library。（David Karnok也是项目Reactor的提交者）

Akka Streams

Akka Streams还彻底实现了Reactive Streams规范。Akka使用Actors处理流数据。虽然Akka Streams符合Reactive Streams API规范，但Akka Streams API与Reactive Streams接口彻底分离。Akka Streams被认为是reactive library。

Ratpack

Ratpack是一组用于构建现代高性能HTTP应用程序的Java库。Ratpack使用Java 8，Netty和Reactive原则。Ratpack提供了Reactive Stream API的基本实现，但并非一个功能齐全的reactive工具包。（可选）您能够将RxJava或Reactor与Ratpack一块儿使用。

Vert.x

Vert.x是一个Eclipse Foundation项目。它是JVM的多语言事件驱动的应用程序框架。Vert.x中的反应支持与Ratpack相似。Vert.x容许您使用RxJava或其本机实现的Reactive Streams API。

Reactive Streams and JVM Releases

Reactive Streams for Java 1.8

使用Java 1.8，您将得到对Reactive Streams规范的强大支持。

在Java 1.8中，Reactive stream不是Java API的一部分。可是，它能够做为一个单独的罐子提供。

Reactive Streams Maven Dependency

<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams</artifactId>
    <version>1.0.0</version>
</dependency>

虽然您能够直接包含此依赖项，但您正在使用的任何Reactive Streams实现都应将其自动包含为依赖项。

Reactive Streams for Java 1.9

当你转向Java 1.9时，事情会发生一些变化。Reactive Streams已成为官方Java 9 API的一部分。

您会注意到Reactive Streams接口在Java 9中的Flow类下。但除此以外，API与Jav​​a 1.8中的Reactive Streams 1.0相同。

Conclusion

在Java 9中，Reactive Streams正式成为Java API的一部分。在研究这篇文章时，很明显各类反应性库已经在不断发展和成熟（即David Karnok世代分类）。

在Reactive Streams以前，各类reactive库没法实现互操做性。他们没法互相交谈。早期版本的RxJava与项目Reactor的早期版本不兼容。可是在Java 9发布前夕，主要的反应库采用了Reactive Streams规范。如今，不一样的库能够互操做。

互操做性是一个重要的多米诺骨牌。例如，MongoDB实现了Reactive Streams驱动程序。

如今，在咱们的应用程序中，咱们可使用Reactor或RxJava来使用MongoDB中的数据。咱们仍然在适应Reactive Streams的早期阶段。可是在接下来的一年左右的时间里，咱们能够期待愈来愈多的开源项目提供Reactive Streams兼容性。

我但愿在不久的未来咱们会看到更多的Reactive Streams。这是一个成为Java开发人员的有趣时间！

下节再续！

原文：https://dzone.com/articles/what-are-reactive-streams-in-java

有什么讨论的内容，能够加我公众号：