【发布时间】：2010-05-16 18:59:33
【问题描述】：
对于 Java，是否有任何等效的 Reactive Extensions (.NET)？
关于 Rx（反应式扩展）
Rx 是一个使用可观察集合组成异步和基于事件的程序的库。
我知道来自 JBOSS 的规则引擎，例如 Drools，但还有其他更接近 Microsoft .NET 方法的方法吗？
【问题讨论】：
标签： java .net linq system.reactive
【发布时间】：2010-05-16 18:59:33
【问题描述】：
根据NetFlix blog post，他们已将 Rx 移植到 Java。它似乎还没有发布，但他们确实有发布像开源这样的内部开发工具的历史，所以我希望它最终会发布。
【讨论】：
我不知道有一个 - 坦率地说，很难以如此简洁的方式完成。穿线面很好； Java 在线程方面的能力非常出色......但目前，该语言不支持适当的功能。
LINQ 的一个简洁的设计特点是一切都基于接口，然后使用扩展方法来添加额外的功能。这使得代码可以流畅地阅读——想象一下如果你必须写：
IObservable<int> = Observable.Select(
Observable.Where(
source, x => x.SomeCondition)
x => x.SomeProjection);
咳咳。扩展方法更加优雅：
IObservable<int> = source.Where(x => x.SomeCondition)
.Select(x => x.SomeProjection);
现在，Java 版本可以改为基于抽象基类，但这会失去一些优雅。
接下来是 lambda 表达式和方法组转换——这些确实是一般 LINQ 的基础； Java 中最接近的等价物（匿名内部类）太难看，无法在 C# 中使用 lambda 表达式的任何地方使用。
基本上，Java 版本的 Rx 是可行的，但它不会像 C# 版本那样优雅——据我所知，这可能是它尚未完成的原因。可能还有其他用于 Java 的“异步驱动”库，但它们不太可能像 Rx 那样全面和简洁。
可能Java 7 中的更改将使这更可行；据我所知，没有引入扩展方法，但是如果 lambda 语法最终没问题，抽象基类版本会是合理的......
【讨论】：
-
好的，谢谢。也许这在带有闭包的 Java 7 中更可行。
-
您现在可以用 Java 编写：IObservable
= source.where(condition()).select(selector()).toArray();见 github.com/nicholas22/jpropel-light
如果您仍然感兴趣，我最近一直在研究Java library，它还提供了大多数 Rx 操作符在反应和交互方面。
它绝对不如 .NET 好。 Java没有扩展方法或函数类型，因此，一切都必须通过静态方法和内部类来调用。 在 .NET 中：
Observable.Range(0, 10).SelectMany(
o => Observable.Range(0, o)
).Run(Console.WriteLine);
我的图书馆：
Reactive.run(
Reactive.selectMany(
Reactive.range(0, 10),
new Func1<Observable<Integer>, Integer>() {
public Observable<Integer> invoke(Integer param1) {
return Reactive.range(0, param1);
}
}
), Reactive.println());
Interactive.run(
Interactive.selectMany(
Interactive.range(0, 10),
new Func1<Iterable<Integer>, Integer>() {
public Iterable<Integer> invoke(Integer param1) {
return Interactive.range(0, param1);
}
}
), Interactive.println());
我将 Func0、Func1、Func2 类型用于 lambda，将 Closeable 类型用于注销。一旦 Java 获得一等函数公民，这些 FuncX 类型就可以简单地替换。 Java 7 的 try-with-resources 增强功能可以使用 Closeable。第三个丑陋之处在于各个地方的弱类型推断。
【讨论】：
RxJava（由 Netflix 移植的 RX）在此处可用且稳定： https://github.com/ReactiveX/RxJava.
它甚至可以很好地使用 Java 8 的 lambda 语法！ 你可以这样做：
Observable
.from(Arrays.asList(1, 2, 3, 4, 5))
.filter(val -> val % 2 == 0)
.map(val -> val * val)
.scan(0, (prev, curr) -> prev + curr)
.subscribe(System.out::println)
很酷吗？
【讨论】：
您可以使用 Project Reactor，它实现了 Reactive Streams 规范。它提供了两个 Publisher 接口：Mono 和 Flux。
Mono 可以发射 0 或 1 个项目，而 Flux 可以通过 onNext 信号发射 0…n 个项目。
涵盖 Project Reactor 的 Java 反应式编程的一个很好的教程是 here。
【讨论】：