资料连接

Android RxJava： 这是一份全面的 操作符 使用汇总 （含详细实例讲解）

1. Merge & Zip操作符: 合并数据源

Android RxJava应用：合并数据源

2. Map & FlapMap & ConcatMap & Buffer: 变换操作符

Android RxJava应用：变换操作符

3. retry & retryUntil & retryWhen : 错误处理操作符

• retry
  retry(): 让被观察者重新发射数据，要是一直错误就一直发送了
  retry(BiPredicate)： interger是第几次重新发送，Throwable是错误的内容
  retry(long time): 最多让被观察者重新发射数据多少次
  retry(long time,Predicate predicate)： 最多让被观察者重新发射数据多少次，在predicate里面进行判断拦截 返回是否继续
  retry(Predicate predicate)： 在predicate里面进行判断拦截 返回是否继续
• retryUntil
  具体使用类似于retry（Predicate predicate），唯一区别：返回 true 则不重新发送数据事件。此处不作过多描述
• retryWhen
• 遇到错误时，将发生的错误传递给一个新的被观察者（Observable），并决定是否需要重新订阅原始被观察者（Observable）& 发送事件
  注意：
  retryWhen 通过 flatMap 返回的 Observable.just(1) 仅仅作为一个信号（同样会被onNext方法捕获），触发了原始的 Observable 重新订阅，从而重新执行 onNext(1) 和 onNext(2)

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onError(new Exception("发生错误了"));
                e.onNext(3);
            }
        })
                // 遇到error事件才会回调
                .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
                    
                    @Override
                    public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
                        // 参数Observable<Throwable>中的泛型 = 上游操作符抛出的异常，可通过该条件来判断异常的类型
                        // 返回Observable<?> = 新的被观察者 Observable（任意类型）
                        // 此处有两种情况：
                            // 1. 若 新的被观察者 Observable发送的事件 = Error事件，那么 原始Observable则不重新发送事件：
                            // 2. 若 新的被观察者 Observable发送的事件 = Next事件 ，那么原始的Observable则重新发送事件：
                        return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                            @Override
                            public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {

                                // 1. 若返回的Observable发送的事件 = Error事件，则原始的Observable不重新发送事件
                                // 该异常错误信息可在观察者中的onError（）中获得
                                 return Observable.error(new Throwable("retryWhen终止啦"));
                                
                                // 2. 若返回的Observable发送的事件 = Next事件，则原始的Observable重新发送事件（若持续遇到错误，则持续重试）
                                 // return Observable.just(1);
                            }
                        });

                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }
                    @Override
                    public void onNext(Integer value) {
                        Log.d(TAG, "接收到了事件"+ value  );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "对Error事件作出响应" + e.toString());
                        // 获取异常错误信息
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "对Complete事件作出响应");
                    }
                });

4. Transformer & Compose 转换符

Transformer 在RxJava中的使用

• 定义
  Transformer 是一个接口，用于创建一个操作符，可以将一个 Observable 转换为另一个 Observable。通常，Transformer 是通过实现 ObservableTransformer<T, R> 接口来创建的。
• 使用场景
  当你需要在多个地方重用相同的操作链时，使用 Transformer 可以将一系列操作封装成一个可复用的单元。
• 代码

public class MyTransformer<T> implements ObservableTransformer<T, T> {
    @Override
    public ObservableSource<T> apply(Observable<T> upstream) {
        return upstream
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .map(value -> value + " transformed");
    }
}

// 使用
observable.compose(new MyTransformer<>()).subscribe(...);