值得收藏的图解Rxjava Operators

本文中通过图解的方式解释Rxjava中复杂的操作符,值得收藏。其中用到的demo地址:RxJava2-Android-Samples

1. buffer

demo:
Observable<List> buffered = getObservable().buffer(2, 3);
第一个参数表示在emit数据之前,Observable需要缓存多少个数据
第二个参数表示每次emit数据之后跳过几个数据。

图示中就是每次buffer 2个数据之后emit,每次emit之后跳过3个数据。

2. concat

demo:
Observable.concat(aObservable, bObservable)
第一个参数为第一个Observable
第二个参数为第二个Observable
图示中连接两个Observable之后,数据会连接起来,emit a1, a2, a3,b1,b2,b3

3. debounce

demo:
getObservable() .debounce(500, TimeUnit.MILLISECONDS)
第一个参数是时间间隔
第二个参数是时间单位
debounce表示emit数据之后一定时间内没有其他数据出现才真正emit数据。
图示中emit黄球后,在规定时间内又emit绿球,则黄球不会被emit。

4. defer

defer为每一个observer创建一个ObservableSource,这样当第一个observer订阅之后如果ObservableSource中的数据发生变化,第二个订阅的Observer会得到不同的数据。

demo:

Observable.defer(new Callable<ObservableSource>() {
            @Override
            public ObservableSource call() throws Exception {
                return Observable.just(brand);
            }
        });

demo中可以随时改变brand的值,这样不同的Observer可能会得到不同的值。

5. distinct

distinct可以对 emit 的数据做去重处理
demo:

Observable.just(1, 2, 1, 1, 2, 3, 4 ,6, 4)
                    .distinct() 
                    .subscribe(getObserver());

demo中最后emit的数据只有1,2,3,4,6

6. filter

filter按照一定的规则过滤数据
demo:

        Observable.just(1, 2, 3, 4, 5, 6)
                .filter(new Predicate() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer % 2 == 0;
                    }
                })
                .subscribe(getObserver());

demo中原始数据中奇数会被过滤掉。

7. reduce

reduce 对所有数据进行处理,最终emit一个数据。
demo:

        Flowable observable = Flowable.just(1, 2, 3, 4);

        observable.reduce(50, new BiFunction() {
            @Override
            public Integer apply(Integer t1, Integer t2) {
                return t1 + t2;
            }
        }).subscribe(getObserver());

demo中把 50 + 1 +2 +3 +4 的结果60 emit。

8. interval

demo:
Observable.interval(0, 2, TimeUnit.SECONDS);
interval可以延时一定时间后开始按周期emit数据,emit的数据从0开始一次递增。
第一个参数为第一次emit数据时延时时间
第二个参数为emit数据周期
第三个参数为时间单位

9 .last

如果Observable有数据则只emit最后一个数据,如果没有数据则emit默认数据。

demo:

Observable.just("A1", "A2", "A3", "A4", "A5", "A6").last("A1") // the default item ("A1") to emit if the source ObservableSource is empty
                .subscribe(getObserver());

demo中只emit A6,如果Observable没有数据,则会emit 默认数据A1。

10. map

map可以对数据执行一些操作后再emit出去。
demo:

        getObservable()
                // Run on a background thread
                .subscribeOn(Schedulers.io())
                // Be notified on the main thread
                .observeOn(AndroidSchedulers.mainThread())
                .map(new Function<List, List>() {

                    @Override
                    public List apply(List apiUsers) throws Exception {
                        return Utils.convertApiUserListToUserList(apiUsers);
                    }
                })
                .subscribe(getObserver());

demo中把一个ApiUser list转为 User list 了。

11. merge

merge 与concat不同的是把两个 Observable的数据合成一列数据,就像是从一个Observable emit,但是顺序不一定。
demo:

        final String[] aStrings = {"A1", "A2", "A3", "A4"};
        final String[] bStrings = {"B1", "B2", "B3"};

        final Observable aObservable = Observable.fromArray(aStrings);
        final Observable bObservable = Observable.fromArray(bStrings);

        Observable.merge(aObservable, bObservable)
                .subscribe(getObserver());

demo 中最终emit的数据可能是”A1″, “B1”, “A2”, “A3”, “A4”, “B2”, “B3″,还可能是其他顺序。

12. scan

demo:

        Observable.just(1, 2, 3, 4, 5)
                // Run on a background thread
                .subscribeOn(Schedulers.io())
                // Be notified on the main thread
                .observeOn(AndroidSchedulers.mainThread())
                .scan(new BiFunction() {
                    @Override
                    public Integer apply(Integer int1, Integer int2) throws Exception {
                        return int1 + int2;
                    }
                })
                .subscribe(getObserver());

demo中依次输出1,3,6,10,15,即依次把BiFunction作用在前一个输出结果和当前数据上。

13. skip

demo:

        Observable.just(1, 2, 3, 4, 5)
                // Run on a background thread
                .subscribeOn(Schedulers.io())
                // Be notified on the main thread
                .observeOn(AndroidSchedulers.mainThread())
                .skip(3)
                .subscribe(getObserver());

skip比较简单,会跳过前几个数据,具体可以通过参数设置,demo中是跳过前三个数据。

14. take

demo:

        Observable.just(1, 2, 3, 4, 5)
                // Run on a background thread
                .subscribeOn(Schedulers.io())
                // Be notified on the main thread
                .observeOn(AndroidSchedulers.mainThread())
                .take(3)
                .subscribe(getObserver());

take比较简单,只取前几个数据emit,demo中取前三个数据。

15. throttleLast

throttleLast emit一定周期内的最后一个数据。

demo:

        Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter emitter) throws Exception {
                // send events with simulated time wait
                Thread.sleep(0);
                emitter.onNext(1); // skip
                emitter.onNext(2); // deliver
                Thread.sleep(505);
                emitter.onNext(3); // skip
                Thread.sleep(99);
                emitter.onNext(4); // skip
                Thread.sleep(100);
                emitter.onNext(5); // skip
                emitter.onNext(6); // deliver
                Thread.sleep(305);
                emitter.onNext(7); // deliver
                Thread.sleep(510);
                emitter.onComplete();
            }
        }).throttleLast(500, TimeUnit.MILLISECONDS)
                // Run on a background thread
                .subscribeOn(Schedulers.io())
                // Be notified on the main thread
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(getObserver());

demo中每隔500ms emit当时的最后一个数据,demo中最终emit 2,6,7。

16. timer

timer比较简单,就是延时一定时间emit 数据0。
demo:

        Observable.timer(2, TimeUnit.SECONDS)
                // Run on a background thread
                .subscribeOn(Schedulers.io())
                // Be notified on the main thread
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(getObserver());

17 zip

demo:

    private void doSomeWork() {
        Observable.zip(getCricketFansObservable(), getFootballFansObservable(),
                new BiFunction<List, List, List>() {
                    @Override
                    public List apply(List cricketFans, List footballFans) throws Exception {
                        return Utils.filterUserWhoLovesBoth(cricketFans, footballFans);
                    }
                })
                // Run on a background thread
                .subscribeOn(Schedulers.io())
                // Be notified on the main thread
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(getObserver());
    }

    private Observable<List> getCricketFansObservable() {
        return Observable.create(new ObservableOnSubscribe<List>() {
            @Override
            public void subscribe(ObservableEmitter<List> e) throws Exception {
                if (!e.isDisposed()) {
                    e.onNext(Utils.getUserListWhoLovesCricket());
                    e.onComplete();
                }
            }
        });
    }

    private Observable<List> getFootballFansObservable() {
        return Observable.create(new ObservableOnSubscribe<List>() {
            @Override
            public void subscribe(ObservableEmitter<List> e) throws Exception {
                if (!e.isDisposed()) {
                    e.onNext(Utils.getUserListWhoLovesFootball());
                    e.onComplete();
                }
            }
        });
    }

    public static List filterUserWhoLovesBoth(List cricketFans, List footballFans) {
        List userWhoLovesBoth = new ArrayList();
        for (User cricketFan : cricketFans) {
            for (User footballFan : footballFans) {
                if (cricketFan.id == footballFan.id) {
                    userWhoLovesBoth.add(cricketFan);
                }
            }
        }
        return userWhoLovesBoth;
    }

zip对两个Observable的数据进行BiFunction操作,之后再emit出去。demo中getCricketFansObservable 获取到喜欢cricket 的人,getFootballFansObservable获取到喜欢football的人,最终经过BiFunction之后获取到喜欢两项运动的人。

关于Rxjava的更多operators可以参考其官网,地址:http://reactivex.io/documentation/operators.html

IT文库 » 值得收藏的图解Rxjava Operators
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址