RxJava查漏补缺

浅浅的花香味﹌ 2023-10-10 10:29 7阅读 0赞

#### RxJava概念 ####

标准中的观察者设计模式，一个被观察者，多个观察者，多次注册。  
RxJava是改装的观察者设计模式，一个订阅(注册) 一个观察者。  
**一旦满足 起点 和 终点 这样的需求，都可以使用RxJava来实现。**

##### RxJava创建型操作符 #####

create：使用者自己发射事件  
just 内部自己发射的，单一对象  
fromArray 内部自己发射的，数集对象  
empty：内部自己发射的 ，下游默认是Object，无法发出有值事件，只会发射 onComplete  
range：内部自己发射的，start 1 累加 count 5 最后结果：1 2 3 4 5

##### RxJava变换型操作符 #####

1.map 把上一层Int Int变换String 观察者String类型。  
2.flatMap 把上一层Int Int变换ObservableSource\{还可以再次发射多次事件\} 观察者String类型。 **不排序的**  
3.concatMap 把上一层Int Int变换ObservableSource\{还可以再次发射多次事件\} 观察者Bitmap类型。 **排序的**  
4.groupBy 把上一层Int Int变换String(高端配置电脑) 观察者GroupedObservable类型 \{key=“高端”, 细节再包裹一层\}  
5.buffer 100个事件 Integer .buffer(20) 观察者List==五个集合

groupBy

// 上游
          Observable.just(6000, 7000, 8000, 9000, 10000, 14000)
    
           // 变换
          .groupBy(new Function<Integer, String>() {
        
              @Override
              public String apply(Integer integer) throws Exception {
        
                  return integer > 8000 ? "高端配置电脑" : "中端配置电脑"; // 分组
              }
          })
          // 使用groupBy下游是 有标准的
          .subscribe(new Consumer<GroupedObservable<String, Integer>>() {
        
              @Override
              public void accept(final GroupedObservable<String, Integer> groupedObservable) throws Exception {
        
                  Log.d(TAG, "accept: " + groupedObservable.getKey());
                  // 以上还不能把信息给打印全面，只是拿到了，分组的key
    
                  // 输出细节，还需要再包裹一层
                  // 细节 GroupedObservable 被观察者
                  groupedObservable.subscribe(new Consumer<Integer>() {
        
                      @Override
                      public void accept(Integer integer) throws Exception {
        
                          Log.d(TAG, "accept: 类别：" + groupedObservable.getKey() + "  价格：" + integer);
                      }
                  });
              }
          });

输出结果

accept: 中端配置电脑
    accept: 类别：中端配置电脑  价格：6000
    accept: 类别：中端配置电脑  价格：7000
    accept: 类别：中端配置电脑  价格：8000
    accept: 高端配置电脑
    accept: 类别：高端配置电脑  价格：9000
    accept: 类别：高端配置电脑  价格：10000
    accept: 类别：高端配置电脑  价格：14000

buffer:很多的数据，不想全部一起发射出去，分批次，先缓存到Buffer

// 上游
           Observable.create(new ObservableOnSubscribe<Integer>() {
        
               @Override
               public void subscribe(ObservableEmitter<Integer> e) throws Exception {
        
                   for (int i = 0; i < 100; i++) {
        
                       e.onNext(i);
                   }
                   e.onComplete();
               }
           })
           // 变换 buffer
           .buffer(20)
           .subscribe(new Consumer<List<Integer>>() {
        
               @Override
               public void accept(List<Integer> integer) throws Exception {
        
                   Log.d(TAG, "accept: " + integer);
               }
           });

输出结果

accept: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
    accept: [20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
    accept: [40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59]
    accept: [60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79]
    accept: [80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99]

##### RxJava过滤型操作符 #####

1.filter 如果是false全部都发射给下游，如果是true，全部都不发射给下游。  
2.take ：只有在定时器运行基础上 加入take过滤操作符，才有take过滤操作符的价值。  
3.distinct 过滤重复事件。  
4.elementAt 指定发射事件内容，如果无法指定，有默认的事件。

elementAt

// 上游
            Observable.create(new ObservableOnSubscribe<String>() {
        
                @Override
                public void subscribe(ObservableEmitter<String> e) throws Exception {
        
                    e.onNext("九阴真经");
                    e.onNext("九阳真经");
                    e.onNext("易筋经");
                    e.onNext("神照经");
                    e.onComplete();
                }
            })
            // 过滤操作符
            .elementAt(2, "默认经") // 指定下标输出 事件
            // 订阅
            .subscribe(new Consumer<String>() {
         // 下游
                @Override
                public void accept(String s) throws Exception {
        
                    Log.d(TAG, "accept: " + s);
                }
            });

输出结果

accept: 易筋经

##### RxJava条件型操作符 #####

All: 如同 if 那样的功能 ：全部为true，才是true，只要有一个为false，就是false.  
contains 是否包含  
any 全部为 false，才是false， 只要有一个为true，就是true  
如果使用了条件操作符，下一层，接收的类型 就是条件类型(Boolean)

##### RxJava合并型操作符 #####

两个或者多个 被观察者 合并。  
1.startWait，concatWith ：先创建被观察者，然后再组合其他的被观察者，然后再订阅  
2.concat/merge/zip：直接合并多个被观察者，然后订阅  
细节:  
a:startWait 先执行 startWait括号里面的被观察者  
b:concatWait 后执行 concatWait括号里面的被观察者  
c:concat 是按照顺序依次执行 最多四个被观察者进行合并  
d:merge 并列执行的，（演示并列的执行，所以学了intervalRange） 最多四个被观察者进行合并  
e:zip 需要对应关系 需要对应，如果不对应，会被忽略的， 最多9个被观察者 进行合并

##### RxJava异常处理操作符 #####

RxJava中是不标准的throw new IllegalAccessError(“我要报错了”);RxJava标准的e.onError(XXX);

1.onErrorReturn最先拦截到e.onError并且可以给下游返回一个 标识400, throw new XXX 拦截不到，整个程序奔溃  
2.onErrorResumeNext最先拦截到e.onError并且可以给下游返回一个 被观察者（还可以再次发送）, throw new XXX 拦截不到，整个程序奔溃  
3.onExceptionResumeNext 能在发生异常的时候，扭转乾坤，能够处理 throw new XXX，可以真正的让App不奔溃  
4.retry return false; 代表不去重试 return true; 不停的重试， 可指定重试次数

##### RxJava线程切换 #####

1.异步线程区域  
Schedulers.io() ：代表io流操作，网络操作，文件流，耗时操作  
Schedulers.newThread() ： 比较常规的，普普通通  
Schedulers.computation() ： 代表CPU 大量计算 所需要的线程  
2.AndroidSchedulers.mainThread() ： 专门为Android main线程量身定做的

3.给上游分配多次，只会在第一次切换，后面的不切换了（忽略）  
4.给下游分配多次，每次都会去切换  
5.如果不配置异步线程，上游发一次，下游接收一次，上游发一次，下游接收一次，上游发一次，下游接收一次  
6.配置好异步线程，就是异步的表现(线程抢夺)，可能上游先发送N个，然后下游再接收N个，然后上游再发Y个，下游再接收Z个…  
7.传统下载图片的写法，代码容易东一块西一块，RxJava下载图片，基于事件流编程，一条链子，起点和终点

// RxJava如果不配置，默认就是主线程main
            // 上游
            Observable.create(new ObservableOnSubscribe<String>() {
        
                @Override
                public void subscribe(ObservableEmitter<String> e) throws Exception {
        
                    Log.d(TAG, "上游 subscribe: " + Thread.currentThread().getName());
    
                    e.onNext("");
                }
            }).subscribeOn(Schedulers.io()) // todo 给上游配置异步线程    // 给上游分配多次，只会在第一次切换，后面的不切换了
                    .subscribeOn(AndroidSchedulers.mainThread()) // 被忽略
                    .subscribeOn(AndroidSchedulers.mainThread()) // 被忽略
                    .subscribeOn(AndroidSchedulers.mainThread()) // 被忽略
                    .subscribeOn(AndroidSchedulers.mainThread()) // 被忽略
                    // result: io 异步线程
    
              .observeOn(AndroidSchedulers.mainThread()) // todo 给下游配置 安卓主线程    // 给下游分配多次，每次都会去切换
                    .observeOn(AndroidSchedulers.mainThread()) // 切换一次线程
                    .observeOn(AndroidSchedulers.mainThread()) // 切换一次线程
                    .observeOn(AndroidSchedulers.mainThread()) // 切换一次线程
                    .observeOn(Schedulers.io()) // 切换一次线程
                    // result: io 异步线程
    
                    .subscribe(new Consumer<String>() {
         // 下游简化版
                @Override
                public void accept(String s) throws Exception {
        
                    Log.d(TAG, "下游 subscribe: " + Thread.currentThread().getName());
                }
            });

##### RxJava背压模式 #####

背压模式的由来：  
RxJava1.X的时候，还没有背压模式， 我们的上游不停的发射，我们的下游处理不过来，就会照成内存泄漏  
RxJava2.X之后，增加背压模式，Flowable（解决背压）

###### 什么时候用Observable<—>Observer， 什么使用Flowable<—>Subscriber ？ ######

答：发射的事件，大量的事件(1000个)，并且考虑到下游处理不过来，就需要使用Flowable

BackpressureStrategy.ERROR // todo 上游不停的发射大量事件，下游阻塞了 处理不过来，放入缓存池，如果池子满了，就会抛出异常
    BackpressureStrategy.BUFFER // todo  上游不停的发射大量事件，下游阻塞了 处理不过来，放入缓存池，”等待“下游来接收事件处理
    BackpressureStrategy.DROP //  todo   上游不停的发射大量事件，下游阻塞了 处理不过来，放入缓存池，如果池子满了，就会把后面发射的事件丢弃
     // （1 ~ 5000  池子满了4000，   4001 ~ 5000丢弃）
    BackpressureStrategy.LATEST // todo 上游不停的发射大量事件，下游阻塞了 处理不过来，只存储 128个事件

同步的，没有执行Subscription s.request(), 当上游发射1，下游无法处理（没有执行s.request()），会抛出异常  
异步的，上游不停的发射，可以在r02方法中，s.request(10) 可以取出来给 下游接收事件处理的  
一旦下游处理了一次上游的事件，缓存池 - 1

#####  #####