函数响应式编程

1. 函数式编程是一种编程范式。
2. 常见的编程范式有：命令式编程、函数式编程和逻辑式编程。
3. 面向对象就是一种命令式编程。
4. 函数响应式编程=函数式编程思路+响应式编程。它的作用为极大地简化项目，特别是处理嵌套回调的异步事件、复杂的列表过滤和变换以及处理时间相关的等问题。
5. Android中函数响应式编程主要有两大框架，分别是RxJava以及Google退出的Agera。

RxJava概述

1. Rx的定义：Rx = Observables + LINQ + Schedulers。
   其中，Rx即ReactiveX，是Reactive Extensions的缩写。它是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序。开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流，用Schedulers参数化异步数据流的并发处理。
2. RxJava与AsyncTask和Handler相比的优点：在复杂代码逻辑中，RxJava依旧能保持清晰的逻辑。
3. RxJava的异步操作是通过扩展观察者模式来实现的。
4. RxJava的四个角色：Observable、Observer、Subscriber、Subject。

观察者

创建Subscriber

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                s.request(Long.MAX_VALUE);
            }

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

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

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        };
        Flowable<Integer> flowable = Flowable.range(1, 3);
        flowable.subscribe(subscriber);

日志输出如下：

D/MainActivity: onSubscribe
D/MainActivity: onNext:1
D/MainActivity: onNext:2
D/MainActivity: onNext:3
D/MainActivity: onComplete

• onComplete：事件队列完结。RxJava不仅把每个事件单独处理，还会把它们看作一个队列，当不会再有新的onNext发出时，需要触发onComplete()方法作为完成的标志。
• onError：事件队列异常。在事件处理过程中出现异常时，onError()会被触发，同时队列自动终止，不允许再有事件发出。
• onNext：普通的事件。将要处理的事件添加到事件队列中。
• onSubscribe：1.onSubscribe 是2.x新添加的方法，在发射数据前被调用，相当于1.x的onStart方法；2.参数为 Subscription ，Subscription 可用于向上游请求发射多少个元素，也可用于取笑请求；3.必须要调用Subscription 的request来请求发射数据，不然上游是不会发射数据的。

创建Observer

        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe");
            }

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

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

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

        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        });
        observable.subscribe(observer);

日志输出如下：

D/MainActivity: onSubscribe
D/MainActivity: onNext:1
D/MainActivity: onNext:2
D/MainActivity: onNext:3
D/MainActivity: onComplete

• onComplete：事件队列完结。RxJava不仅把每个事件单独处理，还会把它们看作一个队列，当不会再有新的onNext发出时，需要触发onComplete()方法作为完成的标志。
• onError：事件队列异常。在事件处理过程中出现异常时，onError()会被触发，同时队列自动终止，不允许再有事件发出。
• onNext：普通的事件。将要处理的事件添加到事件队列中。
• onSubscribe：1.onSubscribe 是2.x新添加的方法，在发射数据前被调用，相当于1.x的onStart方法；2.可在里面调用#dispose()中断发射数据。

被观察者

创建Observable

使用create方法

        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                e.onNext(0);
                e.onNext(1);
                e.onNext(2);
                e.onComplete();
            }
        });

使用just方法

         Observable<Integer> observable = Observable.just(1, 2, 3);

使用fromArray方法

        Integer[] array = {1, 2, 3};
        Observable<Integer> observable = Observable.fromArray(array);

使用fromIterable方法

        List<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        Observable<Integer> observable = Observable.fromIterable(list);

订阅

订阅只需要一行代码就可以了，如

        observable.subscribe(observer);