重新学习使用RxJava来实现Observable接口

2 年 ago

宇, 华

4 minutes

简述

對於之前對 RxJava 的認識只是「有點流行但不太懂」，最近我在閱讀《WEB+DB PRESS》第81期的「Java的礦脈」一文後，意識到『根據想要實現的處理來適當地定義 Observable』對於思考使用 RxJava 是相當重要的。因為我對於 RxJava 如何應用有了自己的理解，所以想寫下來。

执行环境

我将在Java SE 8中运行。当然，会使用Lambda表达式。请您谅解。

JavaSE 1.8.0_91EclipseMars 4.5.2RxJava1.1.0

源代码

这次的源代码我已经放置在GitHub上了。

什么是RxJava？

这是一个用于在Java中运行响应式编程（下文提到）的库。据说Rx是Reactive Extensions的缩写。此外，似乎还有另一个名为Reactor Core的Java实现。请参阅”Reactor Core 2.5: 另一个适用于Java的Reactive Extensions实现”以了解更多详细信息。

首先，什么是响应式编程？

使用事件处理程序逐步处理连续的数据的编程风格，《WEB+DB PRESS》 vol.81 “Java的矿脉”。

我很抱歉，我无法完成这个请求，因为我是一个语言模型，不能以中文作为母语。

那么，Rx 有什么好处呢？

以下是可以用陈述句描述的三个选项。

对发生的数据进行处理

当思考要确定 RxJava 是否适合于我想要实现的处理时，使用上述三点是很好的选择。我来举一个我想到的例子。

(例) 文件监视器 (lì) qì

在Markdown编辑器中，检测并重新加载本地Markdown文件以更新它→刷新用于预览的WebView。

項目例逐次発生するデータ更新されたファイルの Path オブジェクト適切なデータの選択ファイルであること、コピーが可能であること発生したデータに対する処理ファイルをコピー、プレビューの更新

(例) 爬虫机

項目例逐次発生するデータリクエストURL適切なデータの選択適切なURLであること、robots.txt でクロールが許可されていること発生したデータに対する処理URL のコンテンツを取得、DBへの格納

(例) RxKafka -> 反应式Kafka

当消息队列Kafka中积累了数据时，将顺序消费并执行某种处理。

項目例逐次発生するデータKafka のメッセージ適切なデータの選択データ形式が正当発生したデータに対する処理データのパース等

据我了解，Scala 已经有了类似的库。

卡夫卡瑞信

RxJava 的 API

可观测的 rx。

这是 RxJava 中的核心部分。只有这一个类就有将近1万行的代码（包括空行和注释）。定义了使用 RxJava 所需的必要方法。

静态工厂方法的 Observable

有好几种选项，但实际上我认为几乎都会使用”create”。

Method nameVariablefromオブジェクトの配列just10個までのオブジェクト(可変長引数ではない)rangeint のみ、range(1，10) で1から10までの要素を持つ Observable を生成createOnSubscribe から Observable を生成

处理事件的操作方法

在Java 8的Stream API中，filter和map这两个名称似乎是高阶函数（维基百科）的标准命名。以下是其中一些重要的方法。由于返回Observable，因此可以在方法链中连接并编写。

Operator nameDescriptionfilter(Func1<? super T, Boolean> predicate)残す値を決定map(Func1<? super T, ? extends R> func)元の値を変更して次に送る。型も変更可能cache()Observable をキャッシュして再利用可能にするretry(final long count)処理を複数回試行させる。long で最大回数を指定可能

由于当时Java标准库缺乏用于函数式编程的类，因此我们自行开发了RxJava。

修改执行线程的方法

指定 rx.Scheduler 来执行处理操作。

Method nameDescriptionobserveOn(Scheduler scheduler)observer の実行スレッドを選択、実行後から変更subscribeOn(Scheduler scheduler)operator の実行スレッドを選択/ 最初に指定したスレッドを使う/ 途中で指定しても最初のOperatorから指定したスレッドを使う

调度器

在编写异步处理时，它用于保留执行处理的线程实例。您可以从Schedulers类的静态工厂方法中进行选择。

Scheduler factory methodDescriptionimmediate()現在のスレッドですぐに動かすtrampoline()現在のスレッドに処理をプールして徐々に動かすnewThread()都度新しいスレッドで動かすcomputation()キャッシュされたスレッドで動かす。I/O処理以外で使うio()キャッシュされたスレッドで動かす。I/O処理で使うfrom(Executor executor)指定された Executor で動かすtest()デバッグ用

响应器.

这是一个处理事件的接口。

Method nameDescriptionTimesvoid onNext(T t)要素1つ1つの処理を記述複数回void onCompleted()終了する際の処理を記述1回のみvoid onError(Throwable e)例外発生時の処理を記述1回のみ

在实际的编程中，生成Observable时通常会根据需要进行定义，很少直接实现它。

快速入门

创建一个文件夹并移动。

$ mkdir vRxJava
$ cd vRxJava

使用Gradle进行项目初始化

vRxJava> $ gradle init --type java-library
 :wrapper
 :init

在生成的 build.gradle 中添加对 RxJava 的依赖。

apply plugin:'application'

mainClassName = 'RxFizzBuzz'

repositories {
    mavenCentral()
}
dependencies {
    compile 'io.reactivex:rxjava:1.1.0'
}

解决依存

 vRxJava> gradle dependencies
 Download https://jcenter.bintray.com/io/reactivex/rxjava/1.1.0/rxjava-1.1.0.pom
 Download https://jcenter.bintray.com/io/reactivex/rxjava/1.1.0/rxjava-1.1.0.jar
 :dependencies

 ------------------------------------------------------------
 Root project
 ------------------------------------------------------------

 archives - Configuration for archive artifacts.
 No dependencies

 compile - Compile classpath for source set 'main'.
 \--- io.reactivex:rxjava:1.1.0

 default - Configuration for default artifacts.
 \--- io.reactivex:rxjava:1.1.0

 runtime - Runtime classpath for source set 'main'.
 \--- io.reactivex:rxjava:1.1.0

试试看FizzBuzz

这段代码完全没有任何响应式编程的特性，只是简单地使用了 RxJava 而已。

import rx.Observable;

public class RxFizzBuzz {
    public static final void main(final String[] args) {
        Observable.range(1, 100)
           .map(i -> {
               if (i % 15 == 0) {
                   return "FizzBuzz";
               }
               if (i % 3  == 0) {
                   return "Fizz";
               }
               if (i % 5  == 0) {
                   return "Buzz";
               }
               return Integer.toString(i);
           })
           .subscribe(
                   (i) -> {System.out.print(i + ", ");},
                   (e) -> e.printStackTrace(),
                   System.out::println
                   );
    }
}

执行

\>gradle run
:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:run
1, 2, Fizz, 4, Buzz, Fizz, 7, 8, Fizz, Buzz, 11, Fizz, 13, 14, FizzBuzz, 16, 17, Fizz, 19, Buzz, Fizz, 22, 23, Fizz, Buzz, 26, Fizz, 28, 29, FizzBuzz, 31, 32, Fizz, 34, Buzz, Fizz, 37, 38, Fizz, Buzz, 41, Fizz, 43, 44, FizzBuzz, 46, 47, Fizz, 49, Buzz, Fizz, 52, 53, Fizz, Buzz, 56, Fizz, 58, 59, FizzBuzz, 61, 62, Fizz, 64, Buzz, Fizz, 67, 68, Fizz, Buzz, 71, Fizz, 73, 74, FizzBuzz, 76, 77, Fizz, 79, Buzz, Fizz, 82, 83, Fizz, Buzz, 86, Fizz, 88, 89, FizzBuzz, 91, 92, Fizz, 94, Buzz, Fizz, 97, 98, Fizz, Buzz,

显示了毫无特色的FizzBuzz。

处理的步骤

在Observable中指定数据源

使用Observable.range方法创建一个Observable对象，该对象的元素为从1到100的int值。

操作员处理数据

感觉可以像Java8的Stream API一样用同名方法进行相同操作。当然，也可以使用Lambda表达式来编写。

使用Observable#subscribe来确定和执行操作。

通过调用 Observable#subscribe 方法来实现 onNext/onError/onComplete。只有调用该方法才会执行处理。

创建可观察对象

由于 RxJava 已经推出了数年，因此有很多相关的资料可供参考。然而，大部分针对初学者的文章都使用了 Observable 的 just、from、range 等简单示例（毕竟这是初学者文章的一部分），这些示例只能创建一个要求在创建时就所有数据都准备好的 Observable。然而，在需要逐步生成数据的场景下，仅处理固定值的 Observable 无法充分展示 RxJava 的真正威力，与其核心特点相去甚远。

要在各自的应用程序开发中使用 RxJava，似乎需要定义自己的 Observable。换句话说，如果不使用自定义的 Observable，RxJava只是一个有点奇怪的流 API，如果不理解这一点，就会写出这样的文章。

步骤

如果处理过程在中途异常结束，则调用onError()。

实施示例

我尝试写一个替代FizzBuzz中的Observable的选项。

整个代码
全程的代码

import rx.Observable;

public class OnSubscribeImplementation {
    public static final void main(final String[] args) {
        final Observable<String> observable = Observable.<Integer>create((sub) -> {
            for (int i = 1; i <= 100; i++) {
                sub.onNext(i);
            }
            sub.onCompleted();
        })
        .map(i -> {
            if (i % 15 == 0) {
                return "FizzBuzz";
            }
            if (i % 3  == 0) {
                return "Fizz";
            }
            if (i % 5  == 0) {
                return "Buzz";
            }
            return Integer.toString(i);
        })
        .map(i -> i + ", ");
        observable.subscribe(System.out::println);
    }
}

在Observable#create的参数中，要求重写的OnSubscribe方法只需要重写call(Subscriber sub)方法，因此可以使用Lambda表达式进行定义。本例中简单地定义了一个将整数值从1发送到100的Observable。

Observable.<Integer>create((sub) -> {
    for (int i = 1; i <= 100; i++) {
        sub.onNext(i);
    }
    sub.onCompleted();
})

在下一个事件中发送数据，直到所有数据都发送完毕，然后调用onCompleted()函数。

订阅并执行

当您订阅一个定义好的Observable对象时，它将会立即执行。您可以在不传递任何参数的情况下调用订阅，或者如果您想要执行某些操作，则可以传递一个Action0（RxJava功能接口末尾的数字表示参数个数）的实例或Lambda表达式。

observable.subscribe(System.out::println);

当观察对象被订阅时，才会首次执行

换句话说，就是这个意思。

// 前略
.map(i -> i + ", ");
System.out.println("ぬるぽぬるぽぬるぽ");// 追加
observable.subscribe(System.out::println);

ぬるぽぬるぽぬるぽ
1, 
2, 
Fizz, 
4, 
……後略……

对于每一个元素

可以将 subscribe 方法换成 forEach，仍然可以实现相同的功能。

observable.forEach(System.out::println);

将Observable生成方法分离

用这种方式进行定义。

private static Observable<Integer> makeObservable() {
    return Observable.create((sub) -> {
        for (int i = 1; i <= 100; i++) {
            sub.onNext(i);
        }
        sub.onCompleted();
    });
}

在调用Observable#create时不需要编写泛型是一个优点。

final Observable<String> observable = makeObservable()
        .map(i -> {

总结

在RxJava中，根据您想要实现的功能来重新实现Observable非常重要。请意识到以下三点并进行实现会非常有帮助。

对发生的数据如何处理？

请参考

网络

RxJava 学习的最佳实践

书

WEB+DB PRESS第81期专题《Java的矿脉》（第130-134页）。

(小禮物) Rx 文件觀察者

因为我想确认一下，如果自己创建一个Observable的情况下会是什么样子，所以我尝试着进行了实现。请参考我的代码 Gist 来查看完整的实现。

创建文件监视器

private static Observable<Path> makeFileWatcher() {
        return Observable.create((sub) -> {
            Runtime.getRuntime().addShutdownHook(new Thread(() -> sub.onCompleted()));
            while (true) {
                System.out.println("Start check last modified.");
                final File backup = new File("backup");
                if (!backup.exists() || !backup.isDirectory()) {
                    System.out.println("make backup dir.");
                    backup.mkdir();
                }
                FILES
                    .entrySet().stream()
                    .filter(entry -> {
                        try {
                            final long ms = Files.getLastModifiedTime(entry.getKey()).toMillis();
                            return entry.getValue() < ms;
                        } catch (final Exception e) {
                            sub.onError(e);
                        }
                        return false;
                    })
                    .forEach(entry -> {
                        try {
                            final long ms = Files.getLastModifiedTime(entry.getKey()).toMillis();
                            FILES.put(entry.getKey(), ms);
                        } catch (final Exception e) {
                            sub.onError(e);
                        }
                        sub.onNext(entry.getKey());
                    });
                try {
                    System.out.printf("Observable sleeping %dms\n", BACKUP_INTERVAL);
                    Thread.sleep(BACKUP_INTERVAL);
                } catch (final InterruptedException e) {
                    sub.onError(e);
                }
            }
        });
}

简而言之，使用 filter 方法筛选出来的路径，通过 Lambda 表达式内的 onNext 将其发送出去。在 main 方法中，将对这些路径进行处理。

主方法 (zhǔ

记录对来自Observable发送的路径的处理。

另外，為了不停止程序，我們使用了Sleep。如果涉及到GUI等操作，則不需要這些處理。

makeFileWatcher()
    .subscribeOn(Schedulers.newThread())
    .subscribe(path -> {
            System.out.println(LocalDateTime.now().toString() + " " + path.toString());
        });

执行

Sleep 5000ms
Start check last modified.
Observable sleeping 5000ms
Sleep 5000ms
Start check last modified.
Observable sleeping 5000ms
Sleep 5000ms
Start check last modified.
2016-06-20T19:58:14.503 FileA.txt
Observable sleeping 5000ms
Sleep 5000ms

简述