前言

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