首先RxJava 需要在 JDK1.8+ 的版本 运行

 implementation 'io.reactivex:rxandroid:1.2.1'
 implementation 'io.reactivex:rxjava:1.3.8'

subscribeOn():

Schedulers.immediate()	直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler
Schedulers.newThread()	总是启用新线程，并在新线程执行操作
Schedulers.io()	I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
Schedulers.computation()	计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
AndroidSchedulers.mainThread()	它指定的操作将在 Android 主线程运行。

RxAndroid 的基本使用
RxAndroidBase();

 /*-----------
     *🖐说明:
     *   RxAndroid 的基本使用
     */
    private void RxAndroidBase() {
        //创建被观察者
        Observable<String> stringObservable
                = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                //执行观察者的方法
                subscriber.onNext("姜开元");
            }
        });

        //创建观察者
        //该方法在异常时,回调
        //一旦这个方法执行了!就代表观察者出了问题,所以onNext() 代码 不会执行了
        // 该方法在所有事件都观察完毕,回调,事件序列结束的标记
        //一旦这个方法执行了,就代表观察结束了,即使下面还有方法也不会执行了
        subscriber = new Subscriber<String>() {
            @Override
            public void onError(Throwable e) {
                Log.d("onNext", "测试该方法(失败了~)");
                //该方法在异常时,回调
                //一旦这个方法执行了!就代表观察者出了问题,所以onNext() 代码 不会执行了
            }
            @Override
            public void onNext(String s) {
                Log.d("onNext", s + "测试该方法(成功是回调)");
            }
            @Override
            public void onCompleted() {
                Log.d("onCompleted", "测试该方法,前面的方法都执行完毕了");
                // 该方法在所有事件都观察完毕,回调,事件序列结束的标记
                //一旦这个方法执行了,就代表观察结束了,即使下面还有方法也不会执行了

            }
        };

        //把观察者 注册到被观察者身上,只要这段代码中存在该代码 Call 方法就是立刻执行
        stringObservable.subscribe(subscriber);

    }

RxAndroid 接收一个集合对象 对其中的数据 进行 一个一个的观察
RxAndroidFrom()

  /*-----------
     *🖐说明:
     *   RxAndroid 接收一个集合对象 对其中的数据 进行 一个一个的观察
     */
    private void RxAndroidFrom() {
        //创建一个集合,并装上数据
        List<Integer> integerList = Arrays.asList(1, 2, 3, 4, 5);
        //创建被观察者
        Observable.from(integerList)
                //被观察者 一旦被创建 就指定了一个观察者 ,如此就不用再后面进行注册绑定了
                //如果用不到 onError,onCompleted 就创建这个方法 Action1
                //其中的call方法 就相当于 onNext
                //--from的对象 是什么类型的 Action1 就是什么类型的
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.d("call", integer + "");
                    }
                });
    }

接收一个集合对象,对整个集合进行操作
RxAndroidJust()

   /*-----------
     *🖐说明:
     *   接收一个集合对象,对整个集合进行操作
     */
    private void RxAndroidJust() {
        //创建一个集合 并且赋值
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
        //创建被观察者
        Observable.just(list)
                .subscribe(new Subscriber<List<Integer>>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(List<Integer> integers) {
                        Log.d("Call", integers + "");
                    }
                });
    }

RxAndroid 接收一个和对象,对其中的数据 进行 一个一个的过滤
RxAndroidFilterStream()

  /*-----------
     *🖐说明:
     *   RxAndroid 接收一个和对象,对其中的数据 进行 一个一个的过滤
     */
    private void RxAndroidFilterStream() {
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);

        //创建被观察者
        Observable.from(list)
                //对集合数据进行过滤,只发送符合条件的事件
                .filter(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        if (integer % 2 == 0) {
                            return true;
                        } else {
                            return false;
                        }
                    }
                })
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.d("Call", "过滤后" + integer);
                    }
                });
   }

接收一个和对象,对其中的数据 进行转换
RxAndroidMapStream()

  /*-----------
     *🖐说明:
     *   接收一个和对象,对其中的数据 进行转换
     */
    private void RxAndroidMapStream() {
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
        //创建被观察者
        Observable.from(list)
                //对集合数据 进行转换
                //参数1:原数据类型
                //参数2:结果的数据类型
                //这个方法同样可以将 流 转换为Bitmap , 文件
                //String 字符串或数据库 转换为Bean类等
                .map(new Func1<Integer, String>() {
                    @Override
                    public String call(Integer integer) {
                        return integer.toString();
                    }
                })
                //.map() 可以再次转换
                //由于 上边的 第二个类型是 String 所以 返回值变成了 String
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(String s) {
                        Log.d("成功", s);
                    }
                });
    }

把拿到的集合 进行拆分观察
RxAndroidFlatMap()

 /*-----------
    *🖐说明:
    *   把拿到的集合 进行拆分观察
    */
    private void RxAndroidFlatMap() {
        List<Integer> list1 = Arrays.asList(1, 2, 3, 4, 5);
        List<Integer> list2 = Arrays.asList(6, 7, 8, 9, 10);
        List<Integer> list3 = Arrays.asList(11, 12, 13, 14, 15);

        //TODO: just这个方法的 泛型可以指定多个 很多很多
        Observable.just(list1,list2,list3)
                .flatMap(new Func1<List<Integer>, Observable<?>>() {
                    //这个操作 减少了 1层for 循环
                    @Override
                    public Observable<?> call(List<Integer> integers) {
                        return Observable.from(integers);
                    }
                })
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object o) {
                        Log.d("成功",o+"");
                    }
                });

    }

销毁

  @Override
    protected void onStop() {
        super.onStop();
        //杀死这个对象 释放内存
        if(subscriber != null && !subscriber.isUnsubscribed()){
            subscriber.unsubscribe();
        }
    }

-----------------------------------

附加一个 JAVA8 环境搭建

在工作空间的build.gradle中配置

// Java8 环境的搭建,所使用的配置
repositories {
    mavenCentral()
}

// Java8 环境的搭建,所使用的配置
dependencies {
    classpath 'me.tatarka:gradle-retrolambda:3.3.1'
}

// Java8 环境的搭建,所使用的配置
allprojects {
    mavenCentral()
}

标注

在app/build.gradle中配置
(如果 报错 提示 你的插件之中 已存在 请将 apply plugin: 'me.tatarka.retrolambda' 删除即可)

// Java8 环境的搭建,所使用的配置
apply plugin: 'me.tatarka.retrolambda'

// Java8 环境的搭建,所使用的配置
compileOptions {
    sourceCompatibility JavaVersion.VERSION_1_8
    targetCompatibility JavaVersion.VERSION_1_8
}

标注