RxJava使用详解

RxJava是一种异步数据处理库,也是一种扩展的观察者模式。对于Android开发者来说,使用RxJava时也会搭配RxAndroid,它是RxJava针对Android平台的一个扩展,用于Android 开发,它提供了响应式扩展组件,使用RxAndroid的调度器可以解决Android多线程问题。

观察者模式

四大要素:Observable(被观察者),Observer (观察者),subscribe (订阅),事件。
观察者订阅被观察者,一旦被观察者发出事件,观察者就可以接收到。
RxJava使用详解_第1张图片
扩展的观察者模式
RxJava使用详解_第2张图片
当事件完成时会回调onComplete(),在完成过程中发生了异常会回调onError(),onError()和onComplete()只会回调一个。

引入依赖

    implementation 'io.reactivex.rxjava3:rxjava:3.1.3'
    implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
        //创建被观察者
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("Hello Uncle Xing");
                emitter.onComplete();
            }
        });
        //创建观察者
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.i(tag, "onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                Log.i(tag, "onNext:" + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.i(tag, "onError:" + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.i(tag, "onComplete");
            }
        };
        //订阅事件
        observable.subscribe(observer);

操作符

创建Observable

create:用于创建Observable

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("Hello Uncle Xing");
                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.i(tag, "onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                Log.i(tag, "onNext:" + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.i(tag, "onError:" + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.i(tag, "onComplete");
            }
        });

just:创建一个Observable并自动调用onNext发射数据,just中传递的参数将直接在Observer的onNext方法中接收到

        Observable.just("Uncle Xing").subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.i(tag, "onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                Log.i(tag, "onNext:" + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.i(tag, "onError:" + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.i(tag, "onComplete");
            }
        });

interval:创建一个按固定时间间隔发射整数序列的Observable,可用作定时器。

        Observable.interval(1000, TimeUnit.MILLISECONDS).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }

            @Override
            public void onNext(@NonNull Long aLong) {
                Log.i(tag, "count:" + aLong); //这里是非主线程,会隔1s打印出0,1,2,3....
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

timer:创建一个Observable,它在一个特定延迟后发射一个值

        Observable.timer(1000, TimeUnit.MILLISECONDS).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                Log.i(tag, "count:" + aLong);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

转换Observable

map:对数据进行变换后,可以返回任意值,对数据的变换是1对1进行的。

        Observable.just(666).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Throwable {
                return integer.toString();
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                Log.i(tag, "map:" + s);
            }
        });

flatMap:对数据变换后,返回ObservableSource对象,可以对数据进行一对多,多对多的变换。

        Observable.just(1, 2, 3, 4, 5).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Throwable {
                return Observable.just(integer.toString());
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                Log.i(tag, "accept:" + s);
            }
        });

buffer:把Observable的数据放进一个数据包裹,然后发射这些数据包裹,而不是一次发射一个值

        Observable.just(1, 2, 3, 4, 5, 6).buffer(3).subscribe(new Consumer<List<Integer>>() {
            @Override
            public void accept(List<Integer> integers) throws Throwable {
                Log.i(tag, integers.toString());
            }
        });

Log会分两次打印,第一次打印 [1, 2, 3],第二次打印 [4, 5, 6]

过滤Observable

distinct:去掉重复数据

        Observable.just(1, 2, 3, 4, 2, 3).distinct().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Integer integer) {
                Log.i(tag, "distinct:" + integer);
            }
 
            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

elementAt:取出指定位置的数据

        Observable.just(1, 2, 3, 4).elementAt(1).subscribe(new MaybeObserver<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onSuccess(@NonNull Integer integer) {
                Log.i(tag, "onSuccess:" + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

filter:对数据进行指定规则的过滤

        Observable.just(1, 2, 3, 4).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Throwable {
                return integer > 1;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i(tag, "filter:" + integer);
            }
        });

组合Observable

zip:通过一个函数将多个Observable的发射物结合到一起,基于这个函数的结果为每个结合体发射单个数据项

        Observable<Integer> observable = Observable.just(10, 20, 30, 40);
        Observable<Integer> observable2 = Observable.just(1, 2, 3);
        Observable.zip(observable, observable2, new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) throws Throwable {
                return integer + integer2;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i(tag, "zip:" + integer);
            }
        });

注意:当其中一个Observable发送数据结束或异常,另外一个也停止发送,所以这里只会打印出11,22,33

merge:合并多个Observable的发射物

        Observable<Integer> observable = Observable.just(10, 20, 30, 40);
        Observable<Integer> observable2 = Observable.just(1, 2, 3);
        Observable.merge(observable, observable2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Throwable {
                Log.i(tag, "merge:" + integer);//会打印出10,20,30,1,2,3
            }
        });

错误处理

  1. onErrorReturn:让Observable遇到错误时发射一个特殊的项并且正常终止
  2. onErrorResumeNext:让Observable在遇到错误时开始发射第二个Observable的数据序列

Schedulers(调度器):解决多线程问题

  1. io():用于I/O操作;
  2. computation():计算工作默认的调度器;
  3. immediate():立即执行,允许立即在当前线程执行你指定的工作;
  4. newThread():创建新线程;
  5. trampoline():顺序处理,按需处理队列,并运行队列的每一个任务。

AndroidSchedulers:RxAndroid提供在Android平台的调度器,指定观察者在主线程。

SubscribeOn用于每个Observable对象,ObserveOn用于每个Observer对象

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
                emitter.onNext(100);
                emitter.onComplete();
                Log.i(tag, "subscribe thread:" + Thread.currentThread().getName());//打印subscribe thread:RxNewThreadScheduler-1
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        Log.i(tag, "onNext thread:" + Thread.currentThread().getName());//打印onNext thread:main
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

管理RxJava的生命周期

在使用RxJava的时候,如果没有及时解除订阅,在退出Activity的时候,异步线程还在执行,对Activity的引用还在,此时就会产生内存泄露问题。

可使用RxLifecycle,传送门:https://github.com/trello/RxLifecycle

引入依赖

    implementation 'com.trello.rxlifecycle4:rxlifecycle:4.0.2'
    implementation 'com.trello.rxlifecycle4:rxlifecycle-components:4.0.2'

让你的Activity继承RxAppCompatActivity,Fragment继承RxFragment,其余类似,然后使用bindUntilEvent或者bindToLifecycle

        Observable.interval(1000, TimeUnit.MILLISECONDS)
                .compose(bindUntilEvent(ActivityEvent.DESTROY)) //当前Activity执行到onDestroy时,Observable取消订阅
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Throwable {
                        Log.i(tag, "accept:" + aLong);
                    }
                });
        Observable.interval(1000, TimeUnit.MILLISECONDS)
                .compose(bindToLifecycle())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Throwable {
                        Log.i(tag, "accept:" + aLong);
                    }
                });

使用bindToLifecycle:
如果Observable在onCreate执行,那么当执行到onDestroy时取消订阅。
如果Observable在onStart执行,那么当执行到onStop时取消订阅。
如果Observable在onResume执行,那么当执行到onPause时取消订阅。

RxJava + Retrofit完成网络请求

public interface MyService {
    @GET("gallery/{imageType}/response")
    Observable<List<String>> getImages(@Path("imageType") String imageType);
}
        Retrofit retrofit = new Retrofit.Builder()
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .baseUrl(BASE_URL)
                .build();
        MyService service = retrofit.create(MyService.class);

        service.getImages("banner")
                .compose(bindToLifecycle())
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<List<String>>() {
                    @Override
                    public void accept(List<String> strings) throws Throwable {
                        //todo
                    }
                });

你可能感兴趣的