[Android]RxJava的简单介绍和基本使用(一)

前言

RxJava以及RxAndroid(https://github.com/ReactiveX/RxAndroid)已经火了一段时间，这里简单的介绍一下它们。RxJava 在 GitHub 主页上的自我介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。实际上，它已经变成了一个专为 Android 而生的响应式第三方库，主要是因为它是第一个全方位可用于 Java 支持的响应式第三方库。RxJava 2 保留了我们需要在 Android 上支持的所有 Java 版本。关键字：响应式编程、异步

归结起来，它由三个主要的部分组成：

1.用于表示数据源的一组类
2.用于侦听数据源的一组类
3.用于修改与合并数据的一组方法
当您开始侦听数据源的时候，它将开始或者停止执行某些操作。您可以将其想象为一种网络请求，除非您开始对网络回应进行侦听，否则的话这个网络请求将永远不会触发。并且如果您在数据源任务完成之前取消了对数据源的订阅，那么它很可能会取消这个网络请求。

这些操作可以是同步的，也可以是异步的，因此您可以构造出一个类似网络请求的操作，它会在某个代码块当中运行，但是却是运行在后台线程之上的。或者，它也可以是完全异步的，就像我们在 Android 中弹出 Activity 或者单击 UI 当中的某个组件，这些都可以被视为异步操作。

通常情况下，一个网络请求只会产生一条回应，但是只要您的 UI 还存在于那里，即便您只订阅了一个按钮，您会发现，按钮的单击事件流却可能是无限的。此外，它们所产生的回应也可能是空的，因此这种类型的数据源实际上只有两种功能：成功或者失败，它不会生成任何的数据。您可以将其视为类似写入数据库或者写入文件的操作。它实际上并不会返回相关的回应或者数据；它要么成功完成、要么失败。这种完成、失败的响应方面是由 RxJava 当中的数据源，结合那些所谓「终端事件 (terminal event)」而构建的。这类似于那些可以正常返回、也可以抛出异常的方法。

这种响应也有可能永远不会终止。让我们回到之前的按钮点击操作示例来，如果我们将按钮点击事件构建为数据源，那么只要您的这个 UI 还存在，那么这个响应永远都不会结束。只有当 UI 消失，并且您取消了该按钮点击事件的订阅之后，才可能会结束。在此之前，它将永远不会进入到完成状态。

这一切相当于实现了传统的观察者模式。我们有某些可以产生数据的组件，然后我们只想知道它们所产生的数据是什么，因此我们所要做的仅仅只是观察而已。所以我们可以添加一个监听器，当事件发生的时候就收到相应的通知。

基本使用

前面啰嗦了一大堆，感觉好像没什么卵用。直接进入正题，对于程序猿来说，代码比文字更容易理解吧。

build.gradle 中请依赖，

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
// Because RxAndroid releases are few and far between, it is recommended you also
// explicitly depend on RxJava's latest version for bug fixes and new features.
compile 'io.reactivex.rxjava2:rxjava:2.0.1'

private void createFlowable(final String str) {
        Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                //onNext方法就是发射一个任务：比如本例子中的开灯动作
                e.onNext(str);
                //事件完成后调用
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER); //支持背压,不丢弃数据的处理方式

        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "---onSubscribe---");
                //必须设置可请求的次数，否则不会执行onNext和onComplete方法
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG, "---onNext-->" + s);
                Toast.makeText(mContext, s, Toast.LENGTH_SHORT).show();

            }

            @Override
            public void onError(Throwable t) {
                Log.e(TAG,"---onError-->",t);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "---onComplete-->the light is on");
            }
        };
        //subscribe：发射源执行这个事件
        flowable.subscribe(subscriber);
    }

createFlowable("turn the light on!");执行

04-20 13:43:08.904 12293-12293/org.jan.rxandroiddemo D/SimpleFunc1Activity: ---onSubscribe---04-20 13:43:08.904 12293-12293/org.jan.rxandroiddemo D/SimpleFunc1Activity: ---onNext-->turn the light on!04-20 13:43:08.944 12293-12293/org.jan.rxandroiddemo D/SimpleFunc1Activity: ---onComplete-->the light is on

Flowable大意是流动性的，我们可以把它想象成某种事件队列的发射源。 创建一个Flowable发射源，假设程序猿加班回家开灯这样的一件事来说，开灯然后灯亮，开灯的动作组合由Flowable来安排执行(subscribe)。 Subscriber 可以想象成一个事件的处理和响应接收器。开灯动作之后，如电流通过灯泡的物理反应，就是在这个对象里的方法里（onNext）处理。

是不是觉得写法有些复杂啊，Rxjava中有个just的方法，会自动根据你的输入参数自行调用onNext方法哦

来一波代码示意

  private void createSimpleFlowable(String params) {
        Flowable.just(params).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG,"---accept-->"+s);
                Toast.makeText(mContext, s, Toast.LENGTH_SHORT).show();
            }
        });
    }

 调用一下 createSimpleFlowable("turn the light off!");

04-20 14:15:37.864 18583-18583/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---accept-->turn the light off!

变换入门

哈哈哈，我们还可以在输入到输出的过程中对数据做变换操作，什么意思？上代码。

      Flowable.just("hello! RxJava2")
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.i(TAG,"Android-jan:"+s);
                    }
                });

04-20 14:37:07.634 15173-15173/org.jan.rxandroiddemo I/SimpleFunc1Activity: Android-jan:hello! RxJava2

操作符

1.map转换，map函数会返回一个被转换后的Flowable对象哦

private void map(String str) {
        Flowable.just(str).map(new Function<String, String>() {
            @Override
            public String apply(String s) throws Exception {
                //这儿我们做点小动作
                return s+"下午3点";
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG,"---accept-->"+s);
                Toast.makeText(mContext, s, Toast.LENGTH_SHORT).show();
            }
        });
    }

调用我的方法：map("2017年4月20日"); 看看打印了什么？其实你心中早有结果。

04-20 15:44:48.834 25851-25851/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---accept-->2017年4月20日下午3点

再来一波，看看能不能双map执行，

/**
     * 把输入参数str的hashcode转换成String 并且打印出来
     * @param str
     */
    private void map2(String str) {
        Flowable.just(str).map(new Function<String, Integer>() {
            @Override
            public Integer apply(String s) throws Exception {
                return s.hashCode();
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return integer.toString();
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG,s);
                Toast.makeText(mContext, s, Toast.LENGTH_SHORT).show();
            }
        });
    }

没错，打印结果是正确的。

2.fromIterable方法，如果我现在手里有一组列表，我要通过Flowable把它射出去 ，我来教你

   List<Integer> list = new ArrayList<>();        list.add(10);
        list.add(15);
        list.add(20);
        list.add(256);
        list.add(1000);
        Flowable.fromIterable(list).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG,"---from->"+integer);
            }
        });

打印结果：

04-20 16:38:16.404 24914-24914/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---from->10
04-20 16:38:16.404 24914-24914/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---from->15
04-20 16:38:16.404 24914-24914/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---from->20
04-20 16:38:16.404 24914-24914/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---from->256
04-20 16:38:16.404 24914-24914/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---from->1000

3.flatMap操作符，感觉跟map有点像，实际上比map厉害了，主要用来破除嵌套结构。先不用着急，咱们可以先确定的事是 它可以把一个 Flowable 转换成另一个 Flowable。

  List<Integer> list = new ArrayList<>();
        list.add(10);
        list.add(15);
        list.add(20);
        list.add(256);
        list.add(1000);

        Flowable.just(list).flatMap(new Function<List<Integer>, Publisher<String>>() {
            @Override
            public Publisher<String> apply(List<Integer> integers) throws Exception {
                //这段代码意思不重要，重要的是，它能帮你转换你的输入参数，并返回一个Publisher发射源:Flowable
                List<String> strs = new ArrayList<String>();
                for(Integer integer:integers){
                    strs.add(String.valueOf(integer));
                }
                return Flowable.fromIterable(strs);
            }
        }).subscribe(new DefaultSubscriber<String>() {
            @Override
            public void onNext(String s) {
                //这里接收的不是Integer类型了，已经处理成String类型
                Log.i(TAG,"---onNext-->"+s);
            }

            @Override
            public void onError(Throwable t) {
                Log.e(TAG,"---onError-->",t);
            }

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

打印结果：

04-20 17:28:39.044 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---onNext-->10
04-20 17:28:39.044 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---onNext-->15
04-20 17:28:39.044 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---onNext-->20
04-20 17:28:39.044 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---onNext-->256
04-20 17:28:39.044 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---onNext-->1000
04-20 17:28:39.044 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: ---onComplete---

如果你没完全理解的话，请转 关于RxJava的from()和 flatMap()方法 

4.filter操作符，过滤的意思，假设我们要过滤一组数组，来个条件吧

  Integer[] array = {1,2,3,5,6,4,0};
        Flowable.fromArray(array).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                //把数组中大于4的元素过滤出来
                return integer.intValue()>4;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG,"filter->integer="+integer);
            }
        });

打印结果：

04-20 17:42:33.764 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: filter->integer=5
04-20 17:42:33.764 3101-3101/org.jan.rxandroiddemo I/SimpleFunc1Activity: filter->integer=6

5.take操作符，限制发射源发射的数量，take(3)意思就是只能发射3次元素

 Integer[] array = {1,2,7,5,6,4,0};
        Flowable.fromArray(array).take(3).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG,"take->integer="+integer);
            }
        });

打印结果：

04-20 17:49:54.784 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: take->integer=1
04-20 17:49:54.784 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: take->integer=2
04-20 17:49:54.784 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: take->integer=7

6.doOnNext方法：如果我们想在订阅者接收到数据前干点事情，比如记录日志。

 Integer[] array = {1,2,3,5,6,4,0};
        Flowable.fromArray(array).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                //在订阅之前可以进行日志记录
                Log.i(TAG,"doOnNext->integer="+integer);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG,"subscribe->integer="+integer);
            }
        });

打印结果：

04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: doOnNext->integer=1
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: subscribe->integer=1
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: doOnNext->integer=2
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: subscribe->integer=2
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: doOnNext->integer=3
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: subscribe->integer=3
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: doOnNext->integer=5
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: subscribe->integer=5
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: doOnNext->integer=6
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: subscribe->integer=6
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: doOnNext->integer=4
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: subscribe->integer=4
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: doOnNext->integer=0
04-20 17:51:57.324 22016-22016/org.jan.rxandroiddemo I/SimpleFunc1Activity: subscribe->integer=0

今天就先写这么些吧，反正就是基础使用，各位看官先消化消化吧。