独眼巨人反应组织了Java 8库的寒武纪爆发

什么是独眼巨人反应？

Lambda表达式和默认方法在Java 8中的出现预示了Java语言十年来最大的结构性变化。在此基础上构建了一些新的很酷的API，例如Stream, Optional, CompletableFuture最终Java开发人员可以以更实用的样式进行编码。尽管这是非常受欢迎的，但对于许多增强功能而言，还远远不够。

Stream, Optional, CompletableFuture都共享相同的抽象结构并遵循相同的规则。但是，API并未就通用方法名称达成共识，没关系提供通用接口。例如， Stream#map / Optional#map变为CompletableFuture#thenApply 。此外，通常在集合中缺少添加到Stream & Optional的功能。 List#map在哪里？

JDK Stream实现的性能很好，完全是懒惰的，并且设计得很好，可以扩展，但是只提供了一个有限的潜在运算符子集（也许受数据并行性的约束）。将具有顺序Stream扩展（称为Seq ）的步进库（例如jOOλ）插入到无效空间中。 Seq添加了许多其他的Streaming运算符。通常，jOOλ添加了许多缺失的功能特征，例如元组。

cyclops-react的核心目标以及添加诸如FutureStreams之类的原始功能，是提供一种将JDK API和第三方功能库结合在一起的机制。 Java 8推出后，出现了寒武纪的酷库爆炸式增长。像Javaslang和Project Reactor这样的库。首先， cyclops -react通过扩展JDK并利用jOOλ ， pCollections和Agrona之类的其他库来做到这一点。这些库又在可能的情况下扩展了JDK接口，以添加诸如持久性集合之类的功能，并等待免费的Many Producer Single Consumer Queue。

除了重用和扩展JDK接口之外，我们的目标是通过利用第三方标准（例如反应流API）和在没有设置标准的情况下构建我们自己的抽象，使开发人员可以轻松地与外部库集成。我们目前关注的集成库是Google的Guava，RxJava，Functional Java，Project Reactor和Javaslang 。我们已经创建了用于包装诸如Stream, Optional & CompletableFuture类的类型的抽象-以前不存在接口，或者以前不可能存在接口。我们之所以选择这些目标，是因为我们在微服务体系结构的生产中使用了Cyclops-react，并且能够利用正确的技术解决问题并将其与我们的其余代码库顺利集成是至关重要的。

cyclops-react是一个功能丰富的大型项目，此外还具有许多集成模块。在下面的文章中，我将介绍一些可用的功能，其特定目的是展示cyclops-react如何帮助将JDK上的所有点连接起来，并进入Java 8开源社区的步伐。

扩展JDK

可能的话，cyclops-react扩展了JDK API。例如， ReactiveSeq添加了用于处理错误，异步处理的功能，并且进行了更多扩展，从而扩展了JDK Stream和jOOλ的Seq。 cyclops-react Collection扩展而不是创建新的collection实现，而是实现并扩展了适当的JDK接口。然后， LazyFutureStream -react LazyFutureStream扩展了ReactiveSeq ，并允许在期货流上进行聚合操作，就好像它是一个简单的Stream一样（这对于异步和LazyFutureStream地处理大量典型的Java I / O操作非常有用）。

ListX扩展了List ，但添加了渴望执行的运算符

ListX<Integer> tenTimes = ListX.of(1,2,3,4).map(i->i*10);

cyclops-react增加了许多运算符供用户探索。例如，我们可以同时在多个集合中应用函数

反应流API充当数据的生产者（发布者）和消费者（订户）之间的天然桥梁。所有独眼巨人反应数据类型都从反应流中实现Publisher接口，并且还提供了可以转换为任何独眼巨人反应类型的Subscriber实现。这使得与其他基于反应流的库（例如Project Reactor）的直接集成变得简单。

例如，我们可以从任何独眼巨人发布者（例如SortedSetX懒惰地填充Reactor Flux，或者从Reactor类型中填充独眼巨人反应类型。

Flux<Integer> stream = Flux.from(SortedSetX.of(1,2,3,4,5,6,7,8));
//Flux[1,2,3,4,5,6,7,8]ListX<Character> list = ListX.fromPublisher(Flux.just("a","b","c"));

Reactor Flux和Mono类型可以直接与cyclops-react For理解一起使用（每个受支持的库在它们的集成模块中也都有它们自己的本机For理解类集）。

// import static com.aol.cyclops.control.For.*;Publishers.each2(Flux.just(1,2,3), i -> ReactiveSeq.range(i,5),Tuple::tuple).printOut();/*
(1, 1)
(1, 2)
(1, 3)
(1, 4)
(2, 2)
(2, 3)
(2, 4)
(3, 3)
(3, 4)
*/

For理解是一种通过级联对适当方法的调用来管理带有flatMap和map方法的类型上的嵌套迭代的方法。在独眼巨人反应中，嵌套语句可以访问先前语句的元素，因此For理解而言，这是管理现有行为的非常有用的方法。例如，为了确保对可能返回空值的现有方法findId和loadData的调用（如果提供了空参数，则将引发NPE），我们可以利用For理解，仅当返回带有值的Optional时，该表达式才能安全地执行loadData来自findId（）

List<Data> data = 
For.optional(findId()).optional(this::loadData);
//loadData is only called if findId() returns a value

类似地，可以使用诸如Try之类的类型来处理来自findId或loadData的异常结果，使用Futures来异步执行链接的方法，依此类推。

建立跨库抽象

Java 8将Monads引入了Java（ Stream, Optional, CompletableFuture ），但没有提供有助于重用的通用接口，实际上， CompletableFuture使用的方法名称与Optional & Stream用于同一功能的方法名称明显不同。因此map成为thenApply ， flatMap thenCompose成为flatMap thenCompose 。在Java 8世界中，单子已成为一种越来越普遍的模式，但是通常没有办法在它们之间进行抽象。在Cyclops-react中，我们没有尝试定义代表monad的接口，而是构建了一组包装器接口和许多自定义适配器，以使跨Java 8的主要功能样式库中的不同实例适应这些包装器。包装器扩展了AnyM （ AnyM缩写，并且有两个子接口– AnyMValue表示解析为单个值的任何AnyMValue类型（例如Optional或CompletableFuture ）或AnyMSeq最终解析为一个值序列（例如Stream）或列表）。独眼巨人扩展包装器提供了一种机制，用于包装来自RxJava，Guava，Reactor，FunctionalJava和Javaslang的类型。

//We can wrap any type from Reactor, RxJava,
//FunctionalJava, Javaslang, Guava
AnyMSeq<Integer> wrapped = Fj.list(List.list(1,2,3,4,5));//And manipulate it
AnyMSeq<Integer> timesTen = wrapped.map(i->i*10);

cyclops-react提供了一组通用接口，这些包装器（和其他cyclops-react类型）都继承自这些接口，从而使开发人员可以编写更多通用的可重用代码。 AnyM扩展了反应流发布者，这意味着您可以使任何Javaslang，Guava，FunctionalJava或RxJava类型成为具有Cyclops-react的反应流发布者。

AnyMSeq<Integer> wrapped = Javaslang.traversable(List.of(1,2,3,4,5));//The wrapped type is a reactive-streams publisher
Flux<Integer> fromJavaslang = Flux.from(wrapped);wrapped.forEachWithError(System.out::println,System.out::err);

此外，来自Cyclops反应的反应性功能直接在AnyM类型上提供。这意味着，例如，我们可以安排从Javaslang或FunctionalJava流中发出数据的时间–延迟或异步执行reduce操作。

AnyMSeq<Integer> wrapped = Javaslang.traversable(Stream.of(1,2,3,4,5));CompletableFuture<Integer> asyncResult = wrapped.futureOperations(Executors.newFixedThreadPool(1)).reduce(50, (acc, next) -> acc + next);
//CompletableFuture[1550]AnyMSeq<Integer> wrapped = FJ.list(list.list(1,2,3,4,5));Eval<Integer> lazyResult = wrapped.map(i -> i * 10).lazyOperations().reduce(50, (acc,next) -> acc + next);
//Eval[15500]HotStream<Integer> emitting = wrapped.schedule("0 * * * * ?", Executors.newScheduledThreadPool(1));emitting.connect().debounce(1,TimeUnit.DAYS).forEachWithError(this::logSuccess,this::logFailure);

在cyclops-react和新的更广泛的Java 8生态系统中，都有很多值得探索的地方，希望您自己玩乐，学习Java 8并扩展Java 8的边界，这将是一次有趣的冒险！