码迷,mamicode.com
首页 > 编程语言 > 详细

RxJava

时间:2019-02-01 18:59:58      阅读:184      评论:0      收藏:0      [点我收藏+]

标签:接收   截断   table   ble   bounce   策略   时间   obj   image   

一 RxJava 的原理
RxJava 有四个基本概念:Observable (可观察者,即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。Observable 和 Observer 通过 subscribe() 方法实现订阅关系,从而 Observable 可以在需要的时候发出事件来通知 Observer。
1.创建一个简单的Observable

public class Observable<T> {
    final OnSubscribe<T> onSubscribe;

    private Observable(OnSubscribe<T> onSubscribe) {
        this.onSubscribe = onSubscribe;
    }
    
    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
        return new Observable<T>(onSubscribe);
    }
public Subscriber subscribe(Subscriber<? super T> subscriber) { // 是 Subscriber 增加的方法 事件还未发送之前执行,用于数据清零或重置 // 如果对线程有要求,要求在主线程上就只能用doOnSubscribe subscriber.onStart(); // 按照设置的执行计划开始执行 onSubscribe.call(subscriber); return subscriber; } // 相当于一个执行计划 public interface OnSubscribe<T> { void call(Subscriber<? super T> subscriber); } }

2. 创建观察者Observer

public interface Observer<T> {
    void onCompleted();
    void onError(Throwable t);
    void onNext(T var1);
}

public abstract class Subscriber<T> implements Observer<T> {
    public void onStart() {
        //可以用于做一些准备工作,例如数据的清零或重置        
        //但如果必须在主线程就不能用了, 只能适用 doOnSubscribe
        System.out.println("准备工作");
    }
}

测试下

// 创建Observable 将Subscriber 订阅上去 执行内部的执行计划  
Observable.create(new Observable.OnSubscribe<Integer>(){
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        for (int i = 0; i <10 ; i++) {
            subscriber.onNext(i);
        }
    }
}).subscribe(new Subscriber<Integer>() {
    ... ...
    @Override
    public void onNext(Integer var1) {
        System.out.println(var1);
    }
});

二 map 操作符
现在想把Integer转换成 字符怎么在中间加步操作呢?

1. 创建一个接口
/**
 * 转换类型
 * @param <T> 转入类型
 * @param <R> 转出类型
 */
public interface Transformer<T, R> {
    R call(T from);
}
//map操作符的作用是将T类型的Event转化成R类型,转化策略抽象成Transformer<T, R>
public <R> Observable<R> map(Transformer<? super T, ? extends R> transformer) {
    return create(new OnSubscribe<R>() { // 生成一个桥接的Observable和 OnSubscribe
        @Override
        public void call(Subscriber<? super R> subscriber) {
            // Observable.this  这个是闭包中上层的Observable对象
            Observable.this.subscribe(new Subscriber<T>() { // 订阅上层的Observable
           // .... ...................省略complete 和error
                @Override
                public void onNext(T var1) {
                    // 将上层的onSubscribe发送过来的Event,通过转换和处理,转发给目标的subscriber
                    subscriber.onNext(transformer.call(var1));
                }
            });
        }
    });
}

测试方法

// 创建第一个Observable
Observable.create(new OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        for (int i = 0; i < 10; i++) {
            subscriber.onNext(i);
        }
    }  
}).map(new Transformer<Integer, String>() {
    @Override
    public String call(Integer from) {
        return "maping " + from;
    } //创建的第二个Observable2
}).subscribe(new Subscriber<String>() {
    @Override
    public void onNext(String var1) {
        System.out.println(var1);
    }
});
//此时被订阅的是Observable2, 内部调用onSubscribe.call(s1),此时的执行计划是Observable2通过map创建的
// onSubscribe.call 里面 根据闭包,创建时Observable2 传入的是Observable1, 所以此时Observable1  开始
// 被订阅, 传入新的Subscriber(内部的onNext 调用的是Observable2的s1.onNext(transForm.call(初始值)))) )

可以吧map 中的逻辑抽取出来

public <R> Observable<R> map2(Transformer<? super T, ? extends R> transformer) {
    return create(new MapOnSubscribe<T, R>(this, transformer));
}

public class MapOnSubscribe<T, R> implements Observable.OnSubscribe<R> {
    final Observable<T> source;
    final Observable.Transformer<? super T, ? extends R> transformer;
    public MapOnSubscribe(Observable<T> source, Observable.Transformer<? super T, ? extends R> transformer) {
        this.source = source;
        this.transformer = transformer;
    }
    @Override
    public void call(Subscriber<? super R> subscriber) {
        source.subscribe(new MapSubscriber<R, T>(subscriber, transformer));
    }
}

public class MapSubscriber<T, R> extends Subscriber<R> {
    final Subscriber<? super T> actual;
    final Observable.Transformer<? super R, ? extends T> transformer;
    public MapSubscriber(Subscriber<? super T> actual, Observable.Transformer<? super R, ? extends T> transformer) {
        this.actual = actual;
        this.transformer = transformer;
    }
//...........省略
    @Override
    public void onNext(R var1) {
        actual.onNext(transformer.call(var1));
    }
}

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        for (int i = 0; i < 10; i++) {
            subscriber.onNext(i);
        }
        subscriber.onCompleted();
    }
}).map2(new Observable.Transformer<Integer, String>() {
    @Override
    public String call(Integer var1) {
        return var1+"demo";
    }
}).subscribe(new Subscriber<Object>() {
    //.........
    @Override
    public void onNext(Object var1) {
        System.out.println(var1);
    }
}


三 flatMap()
flatMap 返回的是一个Observable对象,而 map 返回的是一个普通转换后的对象;
flatMap 返回的Observable对象并不是直接发送到Subscriber的回调中,而是重新创建一个Observable对象,并激活这个Observable对象,使之开始发送事件;而 map 变换后返回的对象直接发到Subscriber回调中;
flatMap 变换后产生的每一个Observable对象发送的事件,最后都汇入同一个Observable,进而发送给Subscriber回调;
map返回类型 与 flatMap 返回的Observable事件类型,可以与原来的事件类型一样;
可以对一个Observable多次使用 map 和 flatMap;
鉴于 flatMap 自身强大的功能,这常常被用于 嵌套的异步操作,例如嵌套网络请求。传统的嵌套请求,一般都是在前一个请求的 onSuccess() 回调里面发起新的请求,这样一旦嵌套多个的话,缩进就是大问题了,而且严重的影响代码的可读性。而RxJava嵌套网络请求仍然通过链式结构,保持代码逻辑的清晰
---------------------
四 lift()

// 更高级的map  将转换逻辑 提取出来
public <R> Observable<R> lift(Operator<? extends R,? super T> operator) {
    return Observable.create(new OnSubscribe<R>() {
        @Override
        public void call(Subscriber<? super R> subscriber) {
            Subscriber newSubscriber = operator.call(subscriber);
            newSubscriber.onStart();
            // 最原始的Onsubscribe 闭包原理
            onSubscribe.call(newSubscriber);
        }
    });
}
public interface Operator<R, T> {
    public Subscriber<? super T> call(Subscriber<? super R> subscriber);
}
.lift(new Operator<String, Integer>() {
    @Override
    public Subscriber<? super Integer> call(final Subscriber<? super String> subscriber) {
        return new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                subscriber.onCompleted();
            }

            @Override
            public void onError(Throwable t) {
                subscriber.onError(t);
            }

            @Override
            public void onNext(Integer var1) {
                subscriber.onNext(var1 + "toString");
            }
        };
    }
});

技术分享图片

 

 

 

 

 

 

 

 

 

 

 

 

RxJava 都不建议开发者自定义 Operator 来直接使用 lift(),而是建议尽量使用已有的 lift() 包装方法(如 map() flatMap() 等)进行组合来实现需求

,因为直接使用 lift() 非常容易发生一些难以发现的错误。

五 Compose
传入一个 可回调的Transformer
六 线程控制

public class Scheduler {
    // 创建线程池
    final Executor executor;
    public Scheduler(Executor executor) {
        this.executor = executor;
    }

    public Worker createWorker() {
        return new Worker(executor);
    }

    //具体任务的执行者
    public static class Worker {
        final Executor executor;
        public Worker(Executor executor) {
            this.executor = executor;
        }

        public void schedule(Runnable runnable) {
            executor.execute(runnable);
        }
    }
}

//创建线程工厂方法

public class Schedulers {
    private static final Scheduler ioScheduler = new Scheduler(Executors.newCachedThreadPool());
    public static Scheduler io() {
        return ioScheduler;
    }
}

实现subscribeOn

public Observable<T> subscribeOn(Scheduler scheduler) {
    return Observable.create(new OnSubscribe<T>() {
        @Override
        public void call(Subscriber<? super T> subscriber) {
            subscriber.onStart();
            // 将事件的生产切换到新的线程。
            scheduler.createWorker().schedule(new Runnable() {
                @Override
                public void run() {
                    Observable.this.onSubscribe.call(subscriber);
                }
            });
        }
    });
}
//subscribeOn是只作用于上层OnSubscribe的,可以让OnSubscribe的call方法在新线程中执行。
Observable.create(new Observable.OnSubscribe<Integer>() {...})
 // subscribeOn 是作用在上层的线程
.subscribeOn(Schedulers.io()).lift(new Operator<Integer, Integer>() {....})
.subscribe(new Subscriber<Integer>() {})

mainlift
注册成功,开始执行
mainsubscribeOn
注册成功,开始执行
pool-1-thread-1OnSubscribe

//所以最后的subscriber 都是在 pool-1-thread-1OnSubscribe  这个线程上执行回调的

实现observerOn

public Observable<T> observeOn(Scheduler scheduler) {
    return Observable.create(new OnSubscribe<T>() {
        @Override
        public void call(Subscriber<? super T> subscriber) {
            subscriber.onStart();
            Scheduler.Worker worker = scheduler.createWorker();
            // 通过创建一个新的跑在新线程上的Subscriber
            Observable.this.onSubscribe.call(new Subscriber<T>() {
                ...........
                @Override
                public void onNext(T var1) {
                    worker.schedule(new Runnable() {
                        @Override
                        public void run() {
                            subscriber.onNext(var1);
                        }
                    });
                }
            });
        }
    });
}

subscribeOn()的线程切换发生在 OnSubscribe 中,即在它通知上一级 OnSubscribe 时,这时事件还没有开始发送,因此 subscribeOn() 的线程控制可以从事件发出的开端就造成影响;即所有的subscriber 都会在该线程上
而 observeOn() 的线程切换则发生在它内建的 Subscriber 中,即发生在它即将给下一级 Subscriber 发送事件时,因此 observeOn() 控制的是它后面的线程

技术分享图片

①和②两处受第一个 subscribeOn() 影响,运行在红色线程;
③和④处受第一个 observeOn() 的影响,运行在绿色线程;
⑤处受第二个 onserveOn() 影响,运行在紫色线程;
而第二个 subscribeOn() ,由于在通知过程中线程就被第一个 subscribeOn() 截断,因此对整个流程并没有任何影响。这里也就回答了前面的问题:当使用了多个 subscribeOn() 的时候,只有第一个 subscribeOn() 起作用

Schedulers.io(): I/O 操作(读写文件、数据库、网络请求等),与newThread()差不多,区别在于io() 的内部实现是是用一个无数量上限的线程池,可以重用空闲的线程,因此多数情况下 io() 效率比 newThread() 更高。值得注意的是,在 io() 下,不要进行大量的计算,以免产生不必要的线程;
Schedulers.newThread(): 开启新线程操作;
//Schedulers.immediate(): 默认指定的线程,也就是当前线程; Rx 2 已经废弃
Schedulers.computation():计算所使用的调度器。这个计算指的是 CPU 密集型计算,即不会被 I/O等操作限制性能的操作,例如图形的计算。这个 Scheduler 使用的固定的线程池,大小为 CPU 核数。值得注意的是,不要把 I/O 操作放在 computation() 中,否则 I/O 操作的等待时间会浪费 CPU;
AndroidSchedulers.mainThread(): RxJava 扩展的 Android 主线程;
六 doOnSubscribe()
之前说过 onStart 是 subScriber比Observer多的一个方法 可以用来初始化数据等
但是由于它是在subscribe()发送 就被调用了,不能指定线程,而是只能执行在 subscribe() 被调用时的线程

doOnSubscribe() 它和 Subscriber.onStart() 同样是在 subscribe() 调用后而且在事件发送前执行,但区别在于它可以指定线程。默认情况下, doOnSubscribe() 执行在 subscribe() 发生的线程;而如果在 doOnSubscribe() 之后有 subscribeOn() 的话,它将执行在离它最近的 subscribeOn() 所指定的线程。
doOnSubscribe()的后面跟一个 subscribeOn() ,就能指定准备工作的线程了

//doOnSubscribe()  内部就是调用lift 返回一个新的Observable 
// 是在调用onSubscribe 的call 方法里 调用operator  , 所以 由于subscribeOn()的线程切换发生在 OnSubscribe 中
//所以它将执行在离它最近的 subscribeOn() 所指定的线程。
public final Observable<T> doOnSubscribe(final Action0 subscribe) {
    return lift(new OperatorDoOnSubscribe<T>(subscribe));
}

RxJava 的使用
from 接收一个集合作为输入,然后每次输出一个元素给subscriber。
just 接收一个可变参数作为输入,最终也是生成数组,调用from(),然后每次输出一个元素给subscriber。
take 最多保留的事件数。
doOnNext 在处理下一个事件之前要做的事。
// 间隔400ms以内的事件将被丢弃 .debounce(400, TimeUnit.MILLISECONDS)
merge 用于合并两个Observable为一个Observable。较为简单。ZIP是将两个Observable 或多个根据传入函数处理合并成一个Observable
concat 顺序执行多个Observable,个数为1 ~ 9。例子稍后与first操作符一起~~
compose 与 flatMap 类似,都是进行变换,返回Observable对象,激活并发送事件。
compose 是唯一一个能够从数据流中得到原始Observable的操作符,所以,那些需要对整个数据流产生作用的操作(比如,subscribeOn()和observeOn())需要使用 compose 来实现。相较而言,如果在flatMap()中使用subscribeOn()或者observeOn(),那么它仅仅对在 flatMap 中创建的Observable起作用,而不会对剩下的流产生影响。这样就可以简化subscribeOn()以及observeOn()的调用次数了。
compose 是对 Observable 整体的变换,换句话说, flatMap 转换Observable里的每一个事件,而 compose 转换的是整个Observable数据流。
flatMap 每发送一个事件都创建一个 Observable,所以效率较低。而 compose 操作符只在主干数据流上执行操作。
建议使用 compose 代替 flatMap
9.timer 可以做定时操作,换句话讲,就是延迟执行。事件间隔由timer控制
10.interval 定时的周期性操作,与timer的区别就在于它可以重复操作。事件间隔由interval控制。
11.throttleFirst 与debounce类似,也是时间间隔太短,就丢弃事件。可以用于防抖操作,比如防止双击。
.throttleFirst(1, TimeUnit.SECONDS)
12. SingleObserver,相当于是精简版。订阅者回调的不是OnNext/OnError/onCompleted,而是回调OnSuccess/OnError。

Subject
Subject这个类,既是Observable又是Observer,啥意思呢?就是它自身既是事件的生产者,又是事件的消费者,相当于自身是一条管道,从一端进,又从另一端出。举个栗子:PublishSubject

OnSubscribe 对象作为参数。OnSubscribe 会被存储在返回的 Observable 对象中,它的作用相当于一个计划表,当 Observable 被订阅的时候,OnSubscribe 的 call() 方法会自动被调用,事件序列就会依照设定依次触发(对于上面的代码,就是观察者Subscriber 将会被调用三次 onNext() 和一次 onCompleted())。这样,由被观察者调用了观察者的回调方法,就实现了由被观察者向观察者的事件传递,即观察者模式。

最后再说下节流的问题
Flow Control(流控)有哪些思路呢?大概是有四种:
背压(Backpressure);
节流(Throttling);
打包处理;
调用栈阻塞(Callstack blocking)
Throttling
sample 操作函数可以指定生产者发射数据的最大速度,多余的数据被丢弃了,在当前时间段内选最后那个值。所以它也叫throttleLast。与之对应的是throttleFirst
打包处理 Collect
如果你不想丢弃数据,则当消费者忙的时候可以使用 buffer 和 window 操作函数来收集数据
buffer和window
最后再来说下背压问题(真形象)
Backpressure
pull 模型 , 消费者等待生产者, 而Rx为push模型,只要生产好就发射出去就会出现背压问题
1.如果没有使用多线程,同步调用时,调用onNext 会阻塞到消费者消费完才执行第二个onNext
2.使用observeOn, 因为是切换了消费者的线程,因此内部实现用队列存储事件。在 Android 中默认的 buffersize 大小是16,因此当消费比生产慢时, 队列中的数目积累到超过16个,就会抛出MissingBackpressureException。
Observable不再支持Backpressure,而是改用Flowable来专门支持Backpressure
BackpressureStrategy.BUFFER 不丢弃数据的处理方式
BackpressureStrategy.DROP
BackpressureStrategy.LATEST
DROP 和LATEST都会丢弃数据。这两种策略相当于一种令牌机制(或者配额机制),下游通过request请求产生令牌(配额)给上游,上游接到多少令牌,就给下游发送多少数据。当令牌数消耗到0的时候,上游开始丢弃数据,而LATEST 是发送上一个接到令牌缓存的最新值。
新版本 Flowable 代替了Observable 强制设置背压策略

Flowable.create(new FlowableOnSubscribe<Integer>() {
    @Override
    public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
        emitter.onNext(1);
    }
}, BackpressureStrategy.BUFFER);

而Consumer即消费者,用于接收单个值,BiConsumer则是接收两个值,Function用于变换对象,Predicate用于判断。这些接口命名大多参照了Java8
在RxJava2.0中,有五种观察者模式:
Observable/Observer
Flowable/Subscriber
Single/SingleObserver
Completable/CompletableObserver
Maybe/MaybeObserver

https://tech.youzan.com/xiang-ying-shi-jia-gou-yu-rxjavazai-you-zan-ling-shou-de-shi-jian/

RxJava

标签:接收   截断   table   ble   bounce   策略   时间   obj   image   

原文地址:https://www.cnblogs.com/mxz1994/p/10342845.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!