RxJava的学习笔记(二)操作符

前言

RxJava的学习笔记(一)基础概念
RxJava的学习笔记(二)操作符
RxJava的学习笔记(三)线程调度
上一节笔记中记录了RxJava2的基础和基本使用,在本节中继续学习RxJava2的操作符使用。

1、创建操作

创建Observable的方法:
  • just( ), 将一个或多个对象转换成发射这个或这些对象的一个Observable
  • fromArray( ),将一个Iterable, 一个Future, 或者一个数组转换成一个Observable
  • repeat( ), 创建一个重复发射指定数据或数据序列的Observable
  • repeatWhen( ),创建一个重复发射指定数据或数据序列的Observable,它依赖于另一个-
    Observable发射的数据
  • create( ),使用一个函数从头创建一个Observable
  • defer( ),只有当订阅者订阅才创建Observable;为每个订阅创建一个新的Observable
  • range( ),创建一个发射指定范围的整数序列的Observable
  • interval( ),创建一个按照给定的时间间隔发射整数序列的Observable
  • timer( ),创建一个在给定的延时之后发射单个数据的Observable
  • empty( ),创建一个什么都不做直接通知完成的Observable
  • error( ),创建一个什么都不做直接通知错误的Observable
  • never( ),创建一个不发射任何数据的Observable
    方法虽然很多,具体以页面展示的功能需求为准,这里举例常用的just,from,fromIterable着几个创建操作符:
just直接传入一个或者多个对象,发射器依次发射传入的对象
    public void justOperator() {
        Observable.just("map", "map1", "map2")
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("accept:" + s);
                    }
                });
    }
fromArray传入一个对象数组
    public void fromArrayOperator() {
        Observable.fromArray(new String[]{"map", "map1", "map2"})
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("accept:" + s);
                    }
                });
    }
fromIterable接收一个 Iterable 容器,例如List
    public void fromIterableOperator() {
        List<Integer> list = new ArrayList<>();
        list.add(10);
        list.add(1);
        list.add(20);

        Observable.fromIterable(list)
                .subscribe(num -> System.out.println("accept:" + num));
    }
注意:下游观察者仅仅只是关心onNext方法,可以在订阅的时候传入Consumer示例:
        //仅仅关心onNext方法。所以可以简写成下面这样
        Consumer consumer = new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                 Log.d(XqTag.TAG,"accept:" + s);
            }
        };

2、变换操作

RxJava的变换操作符有很多,所谓变换,就是将事件序列中的对象或者整个序列进行一定的加工处理,转换成不同的事件或者事件序列。这里演示常用map()、flatMap()、concatMap()、flatMapIterable()这些变换操作符:

map作用是对于上游发射过来的每一个事件都按照指定的函数进行变换,并将结果依次按照同样的序列发射下游。
map.png

示例:

    public static void mapOperator() {
        Observable.just("map", "map1", "map2")
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        return s + "-jltx";
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                         Log.d(XqTag.TAG,"accept:" + s);
                    }
                });
    }

变换后的结果:

08-12 00:26:35.781 17783-17783/cn.jltx.rxjava.rx D/XQTAG: accept:map-jltx
08-12 00:26:35.781 17783-17783/cn.jltx.rxjava.rx D/XQTAG: accept:map1-jltx
08-12 00:26:35.781 17783-17783/cn.jltx.rxjava.rx D/XQTAG: accept:map2-jltx
flatMap作用是可以把一个上游发射过来的一个发射器Observable通过某种方法变换成多个Observables发射器,然后依次将他们所有发射的数据发射到下游的观察者。有一点需要注意它并不能保证事件的顺序。
flatMap.png

示例:

    public static void flatMapOperator() {
        List<Integer> list1 = new ArrayList<>();
        list1.add(1);
        list1.add(2);
        list1.add(3);

        List<Integer> list2 = new ArrayList<>();
        list2.add(4);
        list2.add(5);
        list2.add(6);

        List<Integer> list3 = new ArrayList<>();
        list3.add(7);
        list3.add(8);
        list3.add(9);

        List<Integer> list4 = new ArrayList<>();
        list4.add(10);
        list4.add(11);
        list4.add(12);

        Observable.just(list1,list2,list3,list4)
                .flatMap(new Function<List<Integer>, ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> apply(List<Integer> integers) throws Exception {
                        Log.d(XqTag.TAG,"中间变换apply:" + integers.toString());
                        //使用一个随机的演示验证事件的顺序
                        int delayTime = (int) (1 + Math.random() * 20);
                        return Observable.fromIterable(integers).delay(delayTime, TimeUnit.MILLISECONDS);
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integers) throws Exception {
                         Log.d(XqTag.TAG,"accept:" + integers);
                    }
                });
    }

变换后的结果:

08-11 12:51:39.928 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[1, 2, 3]
08-11 12:51:39.936 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[4, 5, 6]
08-11 12:51:39.937 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[7, 8, 9]
08-11 12:51:39.938 4059-4109/cn.jltx.rxjava.rx D/XQTAG: accept:4
08-11 12:51:39.938 4059-4109/cn.jltx.rxjava.rx D/XQTAG: accept:5
08-11 12:51:39.938 4059-4109/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 12:51:39.939 4059-4108/cn.jltx.rxjava.rx D/XQTAG: accept:1
08-11 12:51:39.940 4059-4108/cn.jltx.rxjava.rx D/XQTAG: accept:2
08-11 12:51:39.940 4059-4108/cn.jltx.rxjava.rx D/XQTAG: accept:3
08-11 12:51:39.940 4059-4059/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[10, 11, 12]
08-11 12:51:39.943 4059-4111/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 12:51:39.943 4059-4111/cn.jltx.rxjava.rx D/XQTAG: accept:11
08-11 12:51:39.944 4059-4111/cn.jltx.rxjava.rx D/XQTAG: accept:12
08-11 12:51:39.944 4059-4110/cn.jltx.rxjava.rx D/XQTAG: accept:7
08-11 12:51:39.948 4059-4110/cn.jltx.rxjava.rx D/XQTAG: accept:8
08-11 12:51:39.948 4059-4110/cn.jltx.rxjava.rx D/XQTAG: accept:9

如上面通过延时delay一个随机事件得到的结果,验证了flatMap产生多个Observable最后合并的最终发射的数据是无序的。

concatMap作用和flatMap的作用一样,都是一个发射器通过某种方法变换成多个Observables发射器,然后依次将他们所有发射的数据发射到下游的观察者。它和flatMap唯一的区别是它自己保证了顺序。
concatMap.png

示例:

public static void concatMapOperator() {
        List<Integer> list1 = new ArrayList<>();
        list1.add(1);
        list1.add(2);
        list1.add(3);

        List<Integer> list2 = new ArrayList<>();
        list2.add(4);
        list2.add(5);
        list2.add(6);

        List<Integer> list3 = new ArrayList<>();
        list3.add(7);
        list3.add(8);
        list3.add(9);

        List<Integer> list4 = new ArrayList<>();
        list4.add(10);
        list4.add(11);
        list4.add(12);

        Observable.just(list1,list2,list3,list4)
                .concatMap(new Function<List<Integer>, ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> apply(List<Integer> integers) throws Exception {
                        Log.d(XqTag.TAG,"中间变换apply:" + integers.toString());
                        //使用一个随机的演示验证事件的顺序
                        int delayTime = (int) (1 + Math.random() * 20);
                        return Observable.fromIterable(integers).delay(delayTime, TimeUnit.MILLISECONDS);
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integers) throws Exception {
                        Log.d(XqTag.TAG,"accept:" + integers);
                    }
                });
    }

变换后的结果:

08-11 13:01:51.461 4319-4319/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[1, 2, 3]
08-11 13:01:51.480 4319-4411/cn.jltx.rxjava.rx D/XQTAG: accept:1
08-11 13:01:51.481 4319-4411/cn.jltx.rxjava.rx D/XQTAG: accept:2
08-11 13:01:51.481 4319-4411/cn.jltx.rxjava.rx D/XQTAG: accept:3
08-11 13:01:51.481 4319-4411/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[4, 5, 6]
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: accept:4
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: accept:5
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 13:01:51.491 4319-4412/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[7, 8, 9]
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: accept:7
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: accept:8
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: accept:9
08-11 13:01:51.511 4319-4413/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[10, 11, 12]
08-11 13:01:51.514 4319-4414/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 13:01:51.514 4319-4414/cn.jltx.rxjava.rx D/XQTAG: accept:11
08-11 13:01:51.514 4319-4414/cn.jltx.rxjava.rx D/XQTAG: accept:12

由以上结果可知cancatMap可以保证顺序

flatMapIterable作用是变换成对的打包数据,然后生成Iterable而不是原始数据和生成的Observables,但是处理方式是相同的。

示例:

public static void flatMapIterableOperator() {
        List<Integer> list1 = new ArrayList<>();
        list1.add(1);
        list1.add(2);
        list1.add(3);

        List<Integer> list2 = new ArrayList<>();
        list2.add(4);
        list2.add(5);
        list2.add(6);

        List<Integer> list3 = new ArrayList<>();
        list3.add(7);
        list3.add(8);
        list3.add(9);

        List<Integer> list4 = new ArrayList<>();
        list4.add(10);
        list4.add(11);
        list4.add(12);

        List<List<Integer>> list = new ArrayList<>();
        list.add(list1);
        list.add(list2);
        list.add(list3);
        list.add(list4);

        Observable.fromIterable(list)
                .flatMapIterable(new Function<List<Integer>, Iterable<Integer>>() {
                    @Override
                    public Iterable<Integer> apply(List<Integer> integers) throws Exception {
                        Log.d(XqTag.TAG,"中间变换apply:" + integers.toString());
                        //使用一个随机的演示验证事件的顺序
                        return integers;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(XqTag.TAG,"accept:" + integer);
                    }
                });
    }

得到的结果是:

08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[1, 2, 3]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:1
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:2
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:3
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[4, 5, 6]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:4
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:5
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[7, 8, 9]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:7
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:8
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:9
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: 中间变换apply:[10, 11, 12]
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:11
08-11 13:25:02.762 5132-5132/cn.jltx.rxjava.rx D/XQTAG: accept:12

3、过滤操作

过滤操作符可用于过滤和选择Observable发射的数据序列。这里演示常用的filter()、take()这几个过滤操作符:

filter作用是过滤数据的,返回false表示拦截此数据。

示例:

    /**
     * filter 是用于过滤数据的,返回false表示拦截此数据。
     */
    public static void filterOperator() {
        Observable.fromArray(1, 10, 6, 0, 20, -1, 8)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer.intValue() > 5;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + integer);
                    }
                });
    }

上面的过滤出数据大于5的所有数据
得到的结果:

08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:10
08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:6
08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:20
08-11 13:36:11.045 5991-5991/cn.jltx.rxjava.rx D/XQTAG: accept:8
take作用是修改上游Observable的行为,只返回前面的N项数据,然后发射完成通知,忽略剩余的数据。也就是只发射发射前面的N项数据。就算上游Observalbe发射的数据少于N项,那么最后下游接收到的数据个数一样跟上游的个数一样,同事不会抛异常,发射完所有的数据后调用onComplete。

示例:

    public static void takeOperator() {
        Observable.fromArray(1, 10, 6, 0, 20, -1, 8)
                .take(10)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe");
                    }

                    @Override
                    public void onNext(Integer value) {
                        Log.d(XqTag.TAG, "onNext:" + value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete");
                    }
                });
    }

得到的结果:

08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:1
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:10
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:6
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:0
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:20
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:-1
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onNext:8
08-11 13:47:35.272 7361-7361/cn.jltx.rxjava.rx D/XQTAG: onComplete

4、组合操作

组合操作一般用于把多个Obserbable按照一定的关系给连接起来,最后在一个公共的订阅者里面进行汇总监听。

merge顾名思义作用是把多个Observalbe提交的结果按照结果的时间顺序结合。一旦某一个的Observable中出现错误就会马上停止合并,并对下游的订阅者执行onError方法。
merge.png

这里使用两个Observable来进行测试,一个被订阅者每隔1000毫秒后开始发射数据,并只取前面的5个数据,它的发射数据顺序为 0 2 4 6 8;另一个被订阅者延时500毫秒后也是每隔1000毫秒发射设局,也是只取前面的5个数据,它的发射数据顺序为1 3 5 7 9 示例:

/**
     * 按照两个Observable提交结果的时间顺序,对Observable进行合并
     */
    public static void mergeOperator() {

        //被订阅者每隔1000毫秒后开始发射数据,并只取前面5个数据 发射顺序为0 2 4 6 8
        Observable observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(5);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据,并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);
        Observable.merge(observable1, observable2).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(Long value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });

    }

得到的结果是:

08-14 10:18:28.657 9735-9735/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-14 10:18:28.659 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:0
08-14 10:18:29.157 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:1
08-14 10:18:29.658 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:2
08-14 10:18:30.160 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:3
08-14 10:18:30.657 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:4
08-14 10:18:31.158 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:5
08-14 10:18:31.657 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:6
08-14 10:18:32.158 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:7
08-14 10:18:32.657 9735-9773/cn.jltx.rxjava.rx D/XQTAG: onNext:8
08-14 10:18:33.157 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onNext:9
08-14 10:18:33.157 9735-9774/cn.jltx.rxjava.rx D/XQTAG: onComplete
mergeDelayError的作用和merge一样,它和merge的唯一区别是就算有一个或者多个Observable出现错误并不会马上停止合并结果,合适其它的结果都合并完成后再最后调用onErro方法。
mergeDelayError.png

例如沿用上面merge的方式改成mergeDelayError并补上一个erroObservable:

public static void mergeDelayErrorOperatorErro() {

        //抛错误的Observable
        Observable errorObservable = Observable.error(new Exception("this is end!")).delay(3500, TimeUnit.MILLISECONDS);

        //被订阅者每隔1000毫秒后开始发射数据,并只取前面5个数据 发射顺序为0 2 4 6 8 第3500毫秒后产生错误
        Observable observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(5).mergeWith(errorObservable);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据,并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);
        Observable.mergeDelayError(observable1, observable2).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(Long value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });
    }

最后得到的结果是:

08-14 10:38:16.197 11015-11015/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-14 10:38:16.197 11015-11081/cn.jltx.rxjava.rx D/XQTAG: onNext:0
08-14 10:38:16.699 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:1
08-14 10:38:17.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:3
08-14 10:38:18.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:5
08-14 10:38:19.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:7
08-14 10:38:20.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onNext:9
08-14 10:38:20.700 11015-11083/cn.jltx.rxjava.rx D/XQTAG: onError:java.lang.Exception: this is end!
Zip操作符的作用是把n个observable提交的结果,严格按照顺序进行合并,最后这n个observable提交的后最终合并的结果的个数跟这n个observable里面发射个数最小那个Observable发射的个数一样。
20150713004523275.png

示例:

public static void zipOperator() {

        //被订阅者每隔1000毫秒后开始发射数据,并只取前面3个数据 发射顺序为0 2 4
        Observable<Long> observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(3);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据,并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable<Long> observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);

        Observable.zip(observable1, observable2, new BiFunction<Long, Long, String>() {
            @Override
            public String apply(Long aLong, Long aLong2) throws Exception {
                return aLong + "+" + aLong2 + "=" + (aLong + aLong2);
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });
    }

得到的结果是:

08-14 11:01:43.685 11940-11940/cn.jltx.rxjava.rx D/XQTAG: onSubscribe
08-14 11:01:44.186 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onNext:0+1=1
08-14 11:01:45.187 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onNext:2+3=5
08-14 11:01:46.187 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onNext:4+5=9
08-14 11:01:46.187 11940-12020/cn.jltx.rxjava.rx D/XQTAG: onComplete

写在最后

写着写着已经深夜了。
最后附上源码:
https://git.oschina.net/jltx/RxJavaRetrofitDemoPro

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 212,686评论 6 492
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,668评论 3 385
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 158,160评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,736评论 1 284
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,847评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 50,043评论 1 291
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,129评论 3 410
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,872评论 0 268
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,318评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,645评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,777评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,470评论 4 333
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,126评论 3 317
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,861评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,095评论 1 267
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,589评论 2 362
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,687评论 2 351

推荐阅读更多精彩内容