反应式编程实战 - RxJava 2.x 基本模式

RxJava 2.x 提供了五种模式，如下所示。

Observable

创建 Observable 非常容易，我们首先需要创建一个 Observable 作为被观察者，然后在创建一个 Observer 作为观察者，然后通过 subscribe() 进行订阅。

public class ObservableDemo {
    public static void main(String[] args) {
        Observable<String> observable = Observable.create(emitter -> {
            emitter.onNext("Hello World");
            emitter.onNext("Hello World");
            emitter.onComplete();
            emitter.onNext("Hello World");
        });

        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("Observer.onSubscribe");
            }
            @Override
            public void onNext(String s) {
                System.out.println("Observer.onNext: " + s);
            }
            @Override
            public void onError(Throwable e) {
                System.out.println("Observer.onError");
            }
            @Override
            public void onComplete() {
                System.out.println("Observer.onComplete");
            }
        };

        observable.subscribe(observer);
    }
}

我们可以使用 create 创建一个 Observable，它拥有 onNext, onError, onCompleted 方法。其中，onNext用于发射数据项，可以多次调用，每调用一次发射一条数据， onError 或 onCompleted 只能调用一次，onError发射错误事件，除非使用 retry() 操作符来截获错误，否则事件流通常会终止。onCompleted 传递一个完成事件，表示不会再发生onNext调用。两者之间互斥，此后就不能再调用该 Observable 的其他方法。

这里，我们也可以改造成链式调用。

public class ObservableDemo2 {
    public static void main(String[] args) {
        Observable.<String>create(emitter -> {
            emitter.onNext("Hello World");
            emitter.onNext("Hello World");
            emitter.onComplete();
            emitter.onNext("Hello World");
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("Observer.onSubscribe");
            }
            @Override
            public void onNext(String s) {
                System.out.println("Observer.onNext: " + s);
            }
            @Override
            public void onError(Throwable e) {
                System.out.println("Observer.onError");
            }
            @Override
            public void onComplete() {
                System.out.println("Observer.onComplete");
            }
        });
    }
}

阅读 RxJava 2.x 源码 io.reactivex.Observable，我们可以知道 subscribe 具有很多重载的方法。有兴趣的读者，可以深入了解下。

我们可以省略 onComplete()，只实现 onNext() 和 onError()。这将不再对 onComplete() 执行任何操作。我们甚至可以忽略 onError()，只指定 onNext()。但是，不实现 onError() 是在生产环境中应该避免的事情。在事件流的任何地方发生的错误都将传播到 onError() 进行处理，然后终止事件流。如果我们没有为 onError() 指定一个操作，那么该错误将不会处理。当然，如果出现错误，我们可以先尝试使用 retry() 操作符恢复并重新订阅可观察到的数据项。

public final Disposable subscribe()
public final Disposable subscribe(Consumer<? super T> onNext)
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) 
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) 
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe)
public final void subscribe(Observer<? super T> observer)

这里，我们简单来了解一下 subscribe(Consumer<? super T> onNext) 的使用吧。

public class ObservableDemo3 {
    public static void main(String[] args) {
        Observable.<String>create(emitter -> {
            emitter.onNext("Hello World");
            emitter.onNext("Hello World");
            emitter.onComplete();
            emitter.onNext("Hello World");
        }).subscribe(System.out::println);
    }
}

注意， onNext, onError, onCompleted 方法不需要直接推送到最终的观察者，它们可以通过 map() 和 filter() 等操作符创建新的 Observable 然后继续发送。

Flowable 是唯一支持背压的模式，它的用法与 Observable 非常相似。（关于背压，笔者会在之后的文章中进行讲解。）

public class FlowableDemo {
    public static void main(String[] args) {
        Flowable.<String>create(e -> {
            e.onNext("Hello world!");
            e.onNext("Hello World");
            e.onComplete();
            e.onNext("Hello World");
        }, BackpressureStrategy.MISSING).subscribe(new Subscriber<String>(){
            @Override
            public void onSubscribe(Subscription subscription) {
                System.out.println("Subscriber.onSubscribe");
            }
            @Override
            public void onNext(String s) {
                System.out.println("Subscriber.onNext: " + s);
            }
            @Override
            public void onError(Throwable throwable) {
                System.out.println("Subscriber.onError");
            }
            @Override
            public void onComplete() {
                System.out.println("Subscriber.onComplete");
            }
        });
    }
}

阅读 RxJava 2.x 源码 io.reactivex.Flowable ，我们可以知道 subscribe 也具有很多重载的方法。

public final Disposable subscribe() 
public final Disposable subscribe(Consumer<? super T> onNext)
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete)
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Subscription> onSubscribe)
public final void subscribe(FlowableSubscriber<? super T> s)
public final void subscribe(Subscriber<? super T> s)

Single

Single 的工作就像 Observable 一样，但是它只有 onSuccess ?事件和 onError 事件，并且它有自己的 SingleObserver 接口。 onSuccess 整合了 onNext 和 onComplete 事件，因此，这里 onSuccess 只能发送一个数据，换句话说，即使多次发送也不会产生效果。

public class SingleDemo {
    public static void main(String[] args) {
        Single.<String>create(e -> {
            e.onSuccess("success");
            e.onSuccess("success");
        }).subscribe(new SingleObserver<String>(){
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("SingleObserver.onSubscribe");
            }
            @Override
            public void onSuccess(String s) {
                System.out.println("SingleObserver.onSuccess:"+s);
            }
            @Override
            public void onError(Throwable e) {
                System.out.println("SingleObserver.onError");
            }
        });
    }
}

从控制台的打印结果可以看出，即使多次发送“success”，但是只会消费一次。

阅读 RxJava 2.x 源码 io.reactivex.Single ，我们可以知道 subscribe 也具有很多重载的方法。

public final Disposable subscribe()
public final Disposable subscribe(final BiConsumer<? super T, ? super Throwable> onCallback)
public final Disposable subscribe(Consumer<? super T> onSuccess)
public final Disposable subscribe(final Consumer<? super T> onSuccess, final Consumer<? super Throwable> onError)
public final void subscribe(SingleObserver<? super T> subscriber)

这里，我们简单来了解一下 subscribe(Consumer<? super T> onSuccess) 的使用吧。

public class SingleDemo2 {
    public static void main(String[] args) {
        Single.<String>create(e -> {
            e.onSuccess("success");
        }).subscribe(System.out::println);
    }
}

我们可以通过 toObservable 转换成一个 Observable 对象。

Single.just("success").toObservable().subscribe(System.out::println);

Completable

Completable 不发送数据，只有 onComplete 事件和 onError 事件。

public class CompletableDemo {
    public static void main(String[] args) {
        Completable.create(e -> {
            e.onComplete();
        })
        .subscribe(System.out::println);
    }
}

此外，我们可以通过 complete() 快速创建一个 Completable 对象，它会立即调用 onComplete 事件。

Completable.complete().subscribe(System.out::println);

或者，也可以通过 fromAction() 或 fromRunnable() 在调用 onComplete 事件之前执行指定的操作。

Completable.fromAction(System.out::println).subscribe();

Maybe

Maybe 结合了 Single 和 Completable 特性。Maybe 包含 onSuccess 、 onError 、 onComplete 事件。 这里， onSuccess 可以发送 0 ~ 1 个数据，换句话说，即使多次发送也不会产生效果。如果调用 onComplete 事件，就会停止发送数据。

public class MaybeDemo {
    public static void main(String[] args) {
        Maybe.<String>create(e -> {
            e.onComplete();
            e.onSuccess("success");
            e.onSuccess("success");
        }).subscribe(System.out::println);
    }
}