RxJava 2.0 简单介绍

Posted 2021-04-30 58无线技术

一年一年又一年，RxJava也新增了2.0版本，那么为什么是新增版本而不说升级版本呢？

因为2.0版本和1.0版本两者并不兼容，2.0版本是基于Reactive-Streams规范重新设计而来；同时1.x版本和2.x版本两者会并行开发维护，但是1.x版本只维护到**2018-03-31**。

下面我们简单介绍一下两者的不同。

0x00 依赖&包名不同

使用rxjava 1.x、2.x版本的依赖如下：

// rxjava 1.xcompile 'io.reactivex:rxjava:1.1.6'// rxjava 2.xcompile "io.reactivex.rxjava2:rxjava:2.x.y"

包名修改如下：

// 1.x -> 2.xrx.** -> io.reactivex.**

0x01 Observable与Flowable

Observable在2.0版本不支持backpressure，它会缓存全部的数据，一一发送给消费者，如果消费不及时，会产生OOM。于此对应，在2.x版本新增了Flowable，支持设置/自定义backpressure，同时在创建时必须制定backpressure。

 Flowable.create(new FlowableOnSubscribe<Object>() {            @Override
            public void subscribe(FlowableEmitter<Object> e) throws Exception {                for (int i = 0; i < 256; i++) {
                    e.onNext(i);
                }
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER).subscribe(System.out::println, Throwable::printStackTrace);

0x02 Single

当使用Single时，生产者调用onSuccess()通知订阅者，同时终止整个事件流，生产者只能发送一个success事件，订阅者也只能收到一个success事件，适用于网络请求等确定只有单个事件的事件流。对于1.x版本而言，则需要主动调用onComplete()来终止事件流。

注意: Single没有onComplete()方法；只能产生success、error两种事件。

  Single.create(s -> s.onSuccess("aaaa"))
                .subscribe(System.out::println, Throwable::printStackTrace);

0x03 Completable

当使用Completable时，生产者通过调用onComplete()终止事件流，订阅者会收到事件结束回调，适用于订阅者仅需要知道事件结束，而不需要执行结果的情形。

注意: Completable没有onSuccess()方法；只能产生complete、error两种事件。

Completable.create(new CompletableOnSubscribe() {            @Override
            public void subscribe(CompletableEmitter e) throws Exception {                // do something;
                e.onComplete();
            }
        }).subscribe();

0x04 Maybe

Maybe是Single和Completable的组合体，相较于Single只能发送一次item，Completable只能通知事件结束，Maybe可以发送最多一个item，也就是可以发送一个item或者直接终止事件流。

当Maybe调用onSuccess()结束事件流时，订阅者收到一次success事件；当Maybe调用onComplete()结束事件流时，订阅者只能收到事件结束事件。

• onSuccess()收到一次事件:

 Maybe.create(new MaybeOnSubscribe<Object>() {            @Override
            public void subscribe(MaybeEmitter<Object> e) throws Exception {
                e.onSuccess("aaa");
            }
        }).subscribe(System.out::println, Throwable::printStackTrace, () -> {
            System.out.println("onCompletable...");
        });

• onComplete()收到结束事件:

 Maybe.create(new MaybeOnSubscribe<Object>() {            @Override
            public void subscribe(MaybeEmitter<Object> e) throws Exception {
                e.onComplete();
            }
        }).subscribe(System.out::println, Throwable::printStackTrace, () -> {
            System.out.println("onCompletable...");
        });

注意: Maybe拥有onSuccess()和onComplete()方法；可以产生success、complete、error三种事件，其中success和complete是对立的。

0x05 Null

2.0x版本不支持传递null事件，会抛出NullPointerException终止整个事件流。

Single.create(new SingleOnSubscribe<Object>() {            @Override
            public void subscribe(SingleEmitter<Object> e) throws Exception {
                e.onSuccess(null);
            }
        }).subscribe(System.out::println, Throwable::printStackTrace);

错误日志如下:

java.lang.NullPointerException: onSuccess called with null. Null values are generally not allowed in 2.x operators and sources.

0x06 取消订阅

1. 接口改变

2.x版本由于按照Reactive-Streams规范进行开发，而在Reactive-Streams中已经定义了org.reactivestreams.Subscription接口

package org.reactivestreams;public interface Subscription {    void request(long var1);    void cancel();
}

，而1.x版本也定义了一个rx.Subscription接口

package rx;public interface Subscription {    void unsubscribe();    boolean isUnsubscribed();
}

2. 简单取消订阅

可以看到两个类名一样，但是接口方法并不一样，含义也不相同，所以为了避免歧义，2.x版本中干掉了旧的Subscription，同时使用Disposable接口来替代旧的Subscription。具体代码如下:

// 1.x 调用unsubscribe()方法来取消订阅final rx.Subscription subscription = rx.Observable.just(1, 2, 3).subscribe();
subscription.unsubscribe();// 2.x 调用dispose()方法来取消订阅final Disposable subscriber = Flowable.just(1, 2, 3).subscribe();
subscriber.dispose();

3. 使用Subscriber取消订阅

在1.x版本中，我们调用subscribe()后会返回一个rx.Subscription，我们可以使用它进行操作；在2.x版本中，我们调用subscribe()时，如果传入的是Subscriber，那就返回值是void，需要大家自己保存引用。

// 1.x rx.Subscription subscription = rx.Observable.just(1, 2, 3)
  .subscribe(new rx.Subscriber<Integer>() {            @Override
            public void onNext(Integer integer) {

            }            @Override
            public void onError(Throwable t) {

            }            @Override
            public void onComplete() {

            }
        });
subscription.unsubscribe();// 2.xResourceSubscriber<Integer> resourceSubscriber = new ResourceSubscriber<Integer>() {            @Override
            public void onNext(Integer integer) {

            }            @Override
            public void onError(Throwable t) {              // must dispose;
                  dispose();
            }            @Override
            public void onComplete() {              // must dispose;
                  dispose();
            }
        };// 注意当传入subscriber进行订阅时，返回值是void，所以需要自己保存;Flowable.just(1, 2, 3).subscribe(resourceSubscriber);
resourceSubscriber.dispose();

4. 批量取消订阅

1.x版本使用rx.CompositeSubscription批量取消订阅；2.x版本使用io.reactivex.disposables.CompositeDisposable批量取消订阅。

0x07 Subject & Processor

按照Reactive-Streams规范，Subject是一种行为，既是消费者，同时也是生成者，最终被定义为org.reactivestreams.Processor接口，故而，在1.x版本中的subject，在2.x版本中就变成了processor，并且支持backpressure。同时2.x版本中保留了1.x版本的subject，配合Observable使用，不过也不支持backpressure。如:

// 1.xSubject<Object, Object> subject= new SerializedSubject<>(PublishSubject.<Object>create());
subject.onNext("aaa");
subject.onError("aaa");
subject.onComplete();// 2.xfinal FlowableProcessor<Object> objectFlowableProcessor =
            PublishProcessor.create().toSerialized();
objectFlowableProcessor.onNext("aa");
objectFlowableProcessor.onError(new Throwable());
objectFlowableProcessor.onComplete();

参考

1. RxJava 2.0