RxJava2初探

1.RxJava概念及原理

RxJava –

Reactive Extensions for the JVM – a library for composing asynchronous

and event-based programs using observable sequences for the Java VM.

为JVM设计使用可观测序列组成异步基于事件的程序库，核心使用观察者模式，以链式函数调用形式展示

（源码地址、wiki ）

2.x相对于1.x的改动

操作符基本没有变化，Action命名变化

Rx1.0-----------Rx2.0

Action1--------Action

Action1--------Consumer

Action2--------BiConsumer

后面的Action都去掉了，只保留了ActionN

Function和RxJava1.0相比，他们都增加了throws Exception，也就是说，在这些方法做某些操作就不需要try-catch。

Flowable.just("file.txt").map(name->Files.readLines(name)).subscribe(lines->System.out.println(lines.size()),Throwable::printStackTrace);

把支持背压的Observable单独抽出来到类Flowable

Scheduler中移除了Schedulers.immediate()这个线程环境，io.reactivex.Scheduler这个抽象类支持直接调度自定义线程任务

网友指出的rxjava1到2升级踩坑实战，不过官方提供了1.x到2.x的迁移库（RxJava2Interop ）

2.操作符、背压、Scheduler、Hook以及结合Lambda

2.1 操作符

2.1.1 subscribeOn与ObserveOn

此处以subscribeOn为例

subscribeOn.png

2.1.2 map、flatMap、switchMap、concatMap

此处以flatMap为例

涉及到的类图

flatMap.png

flatMap:无法保证事件队列U是按序排列

flatMap_chart.png

试例:

private void testFlatMap(){
        Log.d(TAG, "testFlatMap start");
        Observable.just("A", "B", "C", "D", "E").flatMap(new Function<String, Observable<String>>() {
            @Override
            public Observable<String> apply(String s) throws Exception {
                Log.d(TAG,"testFlatMap apply "+s);
                return Observable.just(s + "1",s + "2").subscribeOn(Schedulers.newThread());
            }
        }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onComplete() {
                Log.d(TAG,"testFlatMap ------>onCompleted()");
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG,"testFlatMap ------>onError()" + e);
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG,"testFlatMap ------>onNext:" + s);
            }
        }/*s -> {Log.d(TAG,"testFlatMap ------>accept:" + s);int i = 1 / 0;} ,e -> Log.d(TAG,"testFlatMap ------>onError()" + e)*/);
        Log.d(TAG, "testFlatMap end");
    }

运行结果：


03-22 19:48:32.188 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap start

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap apply A

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap apply B

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap apply C

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap apply D

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap apply E

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap end

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:A1

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:A2

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:B1

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:B2

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:D1

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:D2

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:E1

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:E2

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:C1

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onNext:C2

03-22 19:48:32.198 2092-2092/com.example.administrator.rxjavademo D/MainActivity: testFlatMap ------>onCompleted()

concatMap:可以保证按序

concatMap_chart.png

草图（非标准类图、非标准时序图，orz...）如下

concatMap.png

试例:

同上换成concatMap

运行结果：


03-22 19:55:31.658 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap start

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap apply A

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap end

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:A1

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:A2

03-22 19:55:31.668 10937-15859/com.example.administrator.rxjavademo D/MainActivity: testConcatMap apply B

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:B1

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:B2

03-22 19:55:31.668 10937-15860/com.example.administrator.rxjavademo D/MainActivity: testConcatMap apply C

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:C1

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:C2

03-22 19:55:31.668 10937-15861/com.example.administrator.rxjavademo D/MainActivity: testConcatMap apply D

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:D1

03-22 19:55:31.668 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:D2

03-22 19:55:31.668 10937-15862/com.example.administrator.rxjavademo D/MainActivity: testConcatMap apply E

03-22 19:55:31.678 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:E1

03-22 19:55:31.678 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onNext:E2

03-22 19:55:31.678 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testConcatMap ------>onCompleted()

switchMap:旧的中间Observable<U>产生的事件序列没有发送完毕，在下一个中间的Observable开始发送后则停止对原事件序列订阅下发

switchMap_chart.png

草图（非标准类图、非标准时序图，orz...）如下

switchMap.png

试例:

同上换成switchMap

运行结果:


03-22 19:53:55.538 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap start

03-22 19:53:55.558 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap apply A

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap apply B

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap apply C

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap apply D

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap apply E

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap end

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap ------>onNext:E1

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap ------>onNext:E2

03-22 19:53:55.568 10937-10937/com.example.administrator.rxjavademo D/MainActivity: testSwitchMap ------>onCompleted()

2.2背压

场景：

当生产者生产事件过慢，消费者消费事件过快，消费者阻塞等待生产者，没有问题

当生产者生产事件过快，而消费者消费事件过慢的情况下，需要额外的缓存开销去容纳生产过剩的事件，而背压提供了异步场景下对生产者流速控制的策略

RxJava2.x中把支持背压的函数分离到Flowable类，Observable类中不支持背压

方式：

1.React pull mode

标准的RxJava是push mode，观察者Subscriber发起订阅后，事件流就从被观察者Flowable中生产并push过来，如何将push mode切换到pull mode的方法是Subscription的request方法

上图为网上copy的，下游文字写错，纠正下

//Flowable需要加主动request的，背压策略
    private void testFlowable(){
        Log.d(TAG,"testFlowable start");
//                Flowable.range(1, 10)
//                .subscribeOn(Schedulers.computation())
//                .map(v -> v * v)
//                .subscribe(i -> Log.d(TAG,"range onNext "+i));

        List<Integer> list = new ArrayList<Integer>();
        for(int i=0 ; i<10; i++){
            list.add(i);
        }
        Flowable.fromIterable(list)
                .subscribeOn(Schedulers.computation())
                .subscribe(/*i -> Log.d(TAG,"fromIterable onNext "+i)*/
                        new Subscriber<Integer>() {
                            Subscription subcription;
                            @Override
                            public void onSubscribe(Subscription s) {
                                Log.d(TAG,"fromIterable onSubscribe "+s);
                                subcription = s;
                                subcription.request(1);//默认实现Long.MAX_VALUE
                            }

                            @Override
                            public void onNext(Integer integer) {
                                Log.d(TAG,"fromIterable onNext "+integer);
                                subcription.request(1);
                            }

                            @Override
                            public void onError(Throwable t) {
                                Log.d(TAG,"fromIterable onError "+t);
                            }

                            @Override
                            public void onComplete() {
                                Log.d(TAG,"fromIterable onComplete");
                            }
                        });
        Log.d(TAG,"testFlowable end");
    }

输出结果：


03-23 16:57:19.718 8740-8740/com.example.administrator.rxjavademo D/MainActivity: testFlowable start
03-23 16:57:19.758 8740-8740/com.example.administrator.rxjavademo D/MainActivity: fromIterable onSubscribe null
03-23 16:57:19.758 8740-8740/com.example.administrator.rxjavademo D/MainActivity: testFlowable end
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 0
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 1
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 2
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 3
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 4
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 5
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 6
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 7
03-23 16:57:19.758 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 8
03-23 16:57:19.768 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onNext 9
03-23 16:57:19.768 8740-9904/com.example.administrator.rxjavademo D/MainActivity: fromIterable onComplete

2.丢弃sample、缓冲buffer window

3.onBackpresureBuffer、onBackPressureDrop

此处牵涉到两个概念

Hot Observable跟Cold Observable

Hot Observable指的是被观察者创建即开始生产事件发送，不等有观察者订阅才发送，如鼠标事件、股票价格 Observable.pulish()可以把cold observable转换为hot observable
Cold Observable指的是一般的被观察者等有观察者订阅才开始生产事件发送，interval range函数创建的都属于该类

总结：

背压只适用于异步场景，并不是说同步场景也需要用背压策略去流速控制

针对flowable事件发布速率过快的场景肯定需要加上request做背压

2.3 Scheduler

2.4 Hook

2.5 Lambada

3.android项目应用举例（结合retrofit）

3.1 先添加rxAndroid依赖（默认会自动引入对应的rxJava包）

dependencies {

compile"com.squareup.retrofit2:retrofit:$parent.ext.retrofitVersion"

compile"com.squareup.retrofit2:adapter-rxjava2:$parent.ext.retrofitVersion"

compile"com.squareup.retrofit2:converter-gson:$parent.ext.retrofitVersion"

compile"com.squareup.retrofit2:converter-scalars:$parent.ext.retrofitVersion"

compile"com.squareup.okhttp3:logging-interceptor:$parent.ext.okhttpVersion"

compile"io.reactivex.rxjava2:rxandroid:$parent.ext.rxAndroidVersion"

}

3.2 retrofit添加callAdapter适配业务接口方法返回类型为Observable

业务接口方法形如


@GET("{headUrl}")

Observable<String> getNewstockCount(

@Path(value ="headUrl",encoded =true) String url,

@QueryMapMap params);

此处Observable<String>一般使用会替换成Observable<T> T指的是接受通用返回body的bean类

rerotfit构造如下


public static Retrofit getRetrofitInstance(){
        OkHttpClient okHttpClient = new OkHttpClient.Builder()
                .addInterceptor(new HttpLoggingInterceptor().setLevel(HttpLoggingInterceptor.Level.BODY))
                .connectTimeout(DEFAULT_TIMEOUT, TimeUnit.SECONDS)
                .writeTimeout(DEFAULT_TIMEOUT, TimeUnit.SECONDS)
                .readTimeout(DEFAULT_TIMEOUT, TimeUnit.SECONDS).build();

        return new Retrofit.Builder().baseUrl("XXX")
                .addCallAdapterFactory(RxJava2CallAdapterFactory.createAsync())
                .addConverterFactory(ScalarsConverterFactory.create()).client(okHttpClient).build();
    }

这里加粗的String一般实际运用根据业务需要指定接受相应body的bean类型

3.3直接使用返回的Observable进行subscribe做相应的链式处理


private void testRetrofit(){
        Log.d(TAG, "testRetrofit start");
        //结合retrofit使用
        NetworkApi networkApi = RetrofitFactory.getRetrofitInstance().create(NetworkApi.class);
        networkApi.getNewstockCount("XXXX",new HashMap<String,Object>()).subscribe(new Observer<String>(){
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onComplete() {
                Log.d(TAG,"onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG,"onError="+e.getLocalizedMessage());
            }

            @Override
            public void onNext(String stringResponse) {
                Log.d(TAG,"onNext="+stringResponse);
            }
        });
        Log.d(TAG, "testRetrofit end");
    }

4.相关开源库（RxBus、RxBinding、frodo、RxLifecycle）

RxBinding

frodo

RxLifecycle

5.优缺点

优点：

1.链式调用让用例流程清晰明了，某种程度上来说也是解耦，不再需要Handler 、new Thread

2.强大的数据变换功能以及线程切换，实际运用项目的时候可以插拔自己项目的线程池策略

缺点：

1.学习成本比较高，需要理解各种操作符，初期使用可能会有函数比较冗长的即视感

2.rxjava库升级带来的项目替换成本

最后编辑于：2017.12.06 05:13:48

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 204,445评论 6赞 478
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 85,889评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 151,047评论 0赞 337
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,760评论 1赞 276
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,745评论 5赞 367
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,638评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,011评论 3赞 398
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,669评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 40,923评论 1赞 299
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,655评论 2赞 321
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,740评论 1赞 330
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,406评论 4赞 320
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 38,995评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,961评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,197评论 1赞 260
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,023评论 2赞 350
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,483评论 2赞 342