函数响应式编程
- 函数式编程是一种编程范式。
- 常见的编程范式有：命令式编程、函数式编程和逻辑式编程。
- 面向对象就是一种命令式编程。
- 函数响应式编程=函数式编程思路+响应式编程。它的作用为极大地简化项目，特别是处理嵌套回调的异步事件、复杂的列表过滤和变换以及处理时间相关的等问题。
- Android中函数响应式编程主要有两大框架，分别是RxJava以及Google退出的Agera。
RxJava概述
- Rx的定义：Rx = Observables + LINQ + Schedulers。
其中，Rx即ReactiveX，是Reactive Extensions的缩写。它是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序。开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流，用Schedulers参数化异步数据流的并发处理。 - RxJava与AsyncTask和Handler相比的优点：在复杂代码逻辑中，RxJava依旧能保持清晰的逻辑。
- RxJava的异步操作是通过扩展观察者模式来实现的。
- 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);
