Java的响应式扩展

Question

是否有类似于.NET的Reactive Extensions的Java等效库？

关于Rx(Reactive Extensions)

Rx是一个使用可观察集合来组合异步和基于事件的程序的库。

我知道像JBOSS的Drools这样的规则引擎，但是否有其他更接近微软.NET方式的方法？

- Timo Westkämper

Answer 1

根据这篇NetFlix博客文章，他们已经将Rx移植到了Java。目前似乎还没有可用版本，但是他们有将内部开发的工具作为开源发布的历史，因此我期待它最终会被发布。

- brucet

Answer 2

我不知道有没有这样的东西——实际上以如此简洁的方式实现难度很大。线程方面没问题；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表达式那样随处可用。

基本上，Rx的Java版本是可行的，但不会像C＃版本那样优雅 - 这可能就是我所知道的为什么还没有完成的原因。可能有其他面向异步编程的Java库存在，但它们不太可能像Rx那样全面和整洁。

Java 7中的更改可能会使其更加可行。据我所知，并没有引入扩展方法，但如果Lambda语法最终好起来的话，基于抽象基类的版本将是合理的...

- Jon Skeet

---

好的，谢谢。也许在Java 7中使用闭包会更可行。 - Timo Westkämper

你现在可以用Java编写以下代码：IObservable<Integer> = source.where(condition()).select(selector()).toArray(); 请参阅github.com/nicholas22/jpropel-light。 - NT_

Answer 3

如果你仍然感兴趣，最近我一直在开发一个Java库，它提供了大部分响应式和交互式方面的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());

我在Lambda表达式中使用Func0、Func1和Func2类型，并使用Closeable类型进行注销。一旦Java成为一等公民函数，这些FuncX类型就可以被简单地替换。Java 7的try-with-resources增强功能可以使用Closeable。第三个丑陋之处是由于各种地方的弱类型推断。

- akarnokd

Answer 4

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)

相当酷吧？

- meddle

Answer 5

当Java 7到来时，您可能考虑将用ActionScript 3编写的raix（wiki）移植到Java 7。该项目（我的项目）是开源的，并包括黑盒单元测试框架的.NET版本，以确保兼容性。

与Java一样，AS3也不支持扩展方法或lambda表达式，因此实现可能类似。话虽如此，由于AVM不支持线程，因此在RxAs中没有锁定代码，因此您需要小心处理竞态条件。

- Richard Szalay

Answer 6

你可以使用实现了反应式流规范的 Project Reactor。它提供了两个 Publisher 接口：Mono 和 Flux。

Mono 可以通过 onNext 信号发出 0 或 1 个项目，而 Flux 可以通过 onNext 信号发出 0 到 n 个项目。

关于 Java 反应式编程并涵盖 Project Reactor 的优秀教程在这里。

- Aleksandar Grujic