RxJava使用系列

RxJava概念与观察者设计模式

    1.起点 和 终点，一旦满足 起点 和 终点 这样的需求，都可以使用RxJava来实现。

    2.标准中的观察者设计模式，一个被观察者 ---- 多个观察者 多次注册。

    3.RxJava是改装的观察者设计模式，一个订阅(注册) 一个观察者。

RxJava上游和下游

    1.上游 Observable 被观察者， 下游 Observer 观察者。

    2.ObservableEmitter<Integer> emitter 发射器 发射事件。

    3.拆分来写的，链式调用。

RxJava创建型操作符

    create：使用者自己发射事件

    just 内部自己发射的，单一对象

    fromArray 内部自己发射的，数集对象

    empty：内部自己发射的 ，下游默认是Object，无法发出有值事件，只会发射 onComplete

    range：内部自己发射的，start 1 累加  count 5    最后结果：1 2 3 4 5

RxJava变换型操作符

上游 -------> 变换操作(往右边流向的时候，进行变换) ----------> 下游

    map 

    flatMap

    concatMap

    groupBy

    buffer

RxJava过滤性型操作符

    filter

    take 定时器 

    distinct 过滤重复的元素

    elementAT   指定下标输出

RxJava条件性型操作符

    all 如同if 全部为true 才是true，只要有一个false，就是false

    contains 包含

    any 同all相反，全部为false才是false，有一个为true就是true

RxJava合并性型操作符

    startWith 两个给被观察者合并，先执行后边的被观察者

    concatWith 两个给被观察者合并,和startWith相反 

    concat 多个合并成一个 最多四个 按照顺序执行

    merge 多个合并成一个 最多四个 并列执行，并发

    zip 需要对应关系，对应执行，如果不对应会被忽略，最多九个被观察者

RxJava异常处理操作符

onErrorReturn 能够接受e.onError 接受异常后，中断接受上游的后续发射的事情 作用 处理记录异常 通知给下一层

onErrorResumeNext  能够接受e.onError 接受异常后，也可以返回异常标识，返回被观察者，被观察者可以再次发射事件给下游

onExceptionResumeNext   能在发生异常的时候(throw Exception)保持程序的健壮，(这种异常一定是可以接受的，才能使用)

retry 重试

RxJava线程切换

异步线程区域 上游

Schedulers.io() 代表io流操作 网络操作 耗时操作

Schedulers.Thread() 比较常规的，普普通通

Schedulers.Comptation()  代表CPU大量计算所需要的线程

给上游分配多次，只会在第一次切换，后面就不切换了（忽略）

下游

AndroidSchedulers.mainThread()  专门为Android main线程量身定做 

给下游切换，每次都会去切换

如果不配置异步线程，上游发一次，下游接受一次

如果配置异步线程，就是异步的表现

传统的写法四分五裂，RxJava基于事件流编程，一条线索下来

RxJava背压模式

Flowable 背压模式 什么情况下使用     如果没有背压模式，上游不停发射线索，下游处理不过来，就会造成内存泄漏

策略

1.Error 上游不停发射消息，下游阻塞，放入缓存池，如果池子满了，抛出异常

2.Buffer 上游不停发射消息，下游阻塞，放入缓存池，如果池子满了，等待下游来处理

3.Drop 上游不停发射消息，下游阻塞，放入缓存池，如果池子满了，把后面的丢弃

4.Latest 上游不停发射消息，下游阻塞，放入缓存池，如果池子满了，只存储128个事件

如果是同步的，下游没有调用request(),会抛异常，而如果是异步的，不执行request(),上游不会等待下游，不会抛出异常，所以外界可以调用

RxJava背压模式  Flowable 讲解

Observable---Observer

Flowable---Subscriber

c的设计和Flowable一致的，只不过Flowable增加了背压模式

Flowable下游，Subscription可以传递给外部使用，request()，事件给下游使用

RxJava+Retrofit(封装了OkHttp)

Retrofit请求网络

Retrofit返回一个结果---Observable

最终的结果是RxJava中的被观察者 上游Observable

RxJava泛型高级进阶

由于RxJava大量使用泛型 

限定   通配符 ? 上限 extends 

                          下限 super

读写模式  上限  不可写  可读

                 下限  可写   不完全可读

RxJava手写系列

RxJava书写create操作符

源码

public class Observable {

public Observable(ObservableOnSubscibe subscibe) {

this.subscibe = subscibe;

    }

private ObservableOnSubscibesubscibe;

    public static Observablecreat(ObservableOnSubscibe subscibe) {

return new Observable(subscibe);

    }

public void subscibe(Observer observer){

        observer.onSubscibe();

        subscibe.subscribe(observer);

    }//调用onSubscibe订阅成功，将observer传入subscibe，将执行subscibe里面的方法

}

public interface ObservableOnSubscibe {

public void subscribe(Observer observer);

}

public interface Observer {

public void onSubscibe();

    public void onNext(T item);

    public void onError(Throwable e);

    public void onComplete();

}

RxJava书写create增加读写操作符

什么时候是读写模式  什么时候是上限、下限模式

在方法参数中，show(Test<? extends Object>) 一定是上限和下限

 在真正使用到泛型中，一定是读写模式

RxJava书写just操作符

just操作符思路和流程和create是完全一致的，使用上的区别主要是create是用户自己发射，自己调用onNext(),而just是内部自己发射。

RxJava书写map操作符

public class Observable {

public Observable(ObservableOnSubscibe subscibe) {

this.subscibe = subscibe;

    }

private ObservableOnSubscibesubscibe;

    public static Observablecreat(ObservableOnSubscibe subscibe) {

return new Observable(subscibe);

    }

public void subscibe(Observer observer){

observer.onSubscibe();

        subscibe.subscribe(observer);

    }

public Observable map(Function function){

MapObservable mapObservable=new MapObservable(function,subscibe);

        return new Observable(mapObservable);

    }//添加map方法，链式调用返回Observable

}

public class MapObservableimplements ObservableOnSubscibe {

private Functionfunction;

    private ObservableOnSubscibesource;//管理上一层

    private ObserverobserverEmitter;//管理下层

    public MapObservable() {

}

@Override

    public void subscribe(Observer observerEmitter) {

this.observerEmitter = observerEmitter;

        MapObserver mapObservable =new MapObserver(observerEmitter, source, function);

        source.subscribe(mapObservable);

    }

public MapObservable(Function function, ObservableOnSubscibe source) {

Log.e("fsfsgsgs","MapObservable");

        this.function = function;

        this.source = source;

    }

class MapObserverimplements Observer {

private ObserverobserverEmitter;//管理下层

        private ObservableOnSubscibesource;//管理上一层

        private Functionfunction;

        public MapObserver(Observer observerEmitter, ObservableOnSubscibe source, Function function) {

this.observerEmitter = observerEmitter;

            this.source = source;

            this.function = function;

        }

@Override

        public void onSubscibe() {

observerEmitter.onSubscibe();

        }

@Override

        public void onNext(T item) {

R type =function.apply(item);

            observerEmitter.onNext(type);

        }

@Override

        public void onError(Throwable e) {

observerEmitter.onError(e);

        }

@Override

        public void onComplete() {

observerEmitter.onComplete();

        }

}

}

public interface Function {

public R apply(T t);

}