RxJava2学习笔记(1)

作为github上star数极高的响应式编程java扩展类库,rxjava是啥就不多说了,网上能查到一堆介绍,下面是一些学习记录:

前提依赖:

compile 'io.reactivex.rxjava2:rxjava:2.1.9'

一、Observable

1.1 hello world

rxjava中的核心思路是“生产者-消费者”模型,生产者的java类通常用xxxEmitter命名,字面意思:发射器,可以想象为一个机关枪,一直biu biu biu的向外发射信息,另一端则是靶子(也就是消费者),在不停的接收。不过要注意的是:rxjava中,能接收子弹的靶子,可以同时有多个。

        Observable<String> observable = Observable.create(emitter -> {
            emitter.onNext("a");
            emitter.onNext("b");
            emitter.onNext("c");
            emitter.onComplete();
        });

        Observer observer1 = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("subscribe=>");
            }

            @Override
            public void onNext(@NonNull String s) {
                System.out.println(s + " ");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println(e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("complete");
            }
        };

        observable.subscribe(observer1);

 输出:

subscribe=>
a 
b 
c 
complete

注:最后一行,也可以改成

observable.subscribe(observer1);
observable.subscribe(observer1);

这样就相当于2个靶子在接子弹了。 上面这是最正统的写法,官方推荐使用链式编程写法:

        Observable.create((ObservableOnSubscribe<String>) emitter -> {
            emitter.onNext("a");
            emitter.onNext("b");
            emitter.onNext("c");
            emitter.onComplete();
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("subscribe=>");
            }

            @Override
            public void onNext(@NonNull String s) {
                System.out.println(s + " ");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println(e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("complete");
            }
        });

1.2 onComplete事件

emitter发送onComplete消息后,挨打的靶子(消费者),就不再继续处理了,不管后面emitter是否还继续发送。

Observable.create((ObservableOnSubscribe<String>) emitter -> {
    emitter.onNext("a");
    emitter.onNext("b");
    emitter.onNext("c");
    emitter.onComplete(); //这里主动通知消费者complete
    System.out.println("complete后,emitter还继续发射...");
    emitter.onNext("d");
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
        System.out.println("subscribe=>");
    }

    @Override
    public void onNext(@NonNull String s) {
        System.out.println(s + " ");
    }

    @Override
    public void onError(@NonNull Throwable e) {
        System.out.println(e.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("complete");
    }
});

 输出:

subscribe=>
a 
b 
c 
complete
complete后,emitter还继续发射...

注:onComplete之后,emitter再次发送的"d",消费者已经不再处理了。

1.3 onError事件

onError即可以在emitter(生产者)端报错,也可以在靶子(消费者)上报错,不管哪一端发生error,消费者就停止处理了。

Observable.create((ObservableOnSubscribe<String>) emitter -> {
    emitter.onNext("a");
    emitter.onError(new Throwable("emitter报了个错!"));
    System.out.println("准备发送c");
    emitter.onNext("c");
    emitter.onComplete();
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
        System.out.println("subscribe=>");
    }

    @Override
    public void onNext(@NonNull String s) {
        System.out.println(s + " ");
    }

    @Override
    public void onError(@NonNull Throwable e) {
        System.out.println(e.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("complete");
    }
});

 输出:

subscribe=>
a 
emitter报了个错!
准备发送c

下面模拟下消费者处理时,发生异常:

Observable.create((ObservableOnSubscribe<String>) emitter -> {
    emitter.onNext("a");
    emitter.onNext("b");
    emitter.onNext("c");
    emitter.onComplete();
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
        System.out.println("subscribe=>");
    }

    @Override
    public void onNext(@NonNull String s) {
        if (s.equals("b")) {
            int a = 0;
            int b = 1;
            System.out.println((b / a));
        }
        System.out.println(s + " ");
    }

    @Override
    public void onError(@NonNull Throwable e) {
        System.out.println("error:" + e.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("complete");
    }
});

输出:

subscribe=>
a
error:/ by zero

1.4 disposable

如果消费者主动dispose()后,相当于就解除了生产者-消费者的关系。

Observable.create((ObservableOnSubscribe<String>) emitter -> {
    emitter.onNext("a");
    emitter.onNext("b");
    System.out.println("准备发送c");
    emitter.onNext("c");
    emitter.onComplete();
}).subscribe(new Observer<String>() {

    Disposable disposable;

    @Override
    public void onSubscribe(@NonNull Disposable d) {
        disposable = d;
        System.out.println("subscribe=>");
    }

    @Override
    public void onNext(@NonNull String s) {
        if (s.equals("b")) {
            disposable.dispose();
        }
        System.out.println(s + " ");
    }

    @Override
    public void onError(@NonNull Throwable e) {
        System.out.println("error:" + e.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("complete");
    }
});

 上面的代码,消费者在遇到b时,主动切断了与生产者的关联,emitter再发送的d,消费者就不处理了,输出:

subscribe=>
a 
b 
准备发送c

1.5 大道至简

如果消费者只关心onNext的处理部分,其它无所谓,上面这一堆代码,可以简化为一行:

Observable.fromArray("a", "b", "c").subscribe(c -> System.out.println(c + " "));

 输出:

a 
b 
c 

最后再来一个示例:把3个单词拼成一句话,而且每个单词处理成“首字母大写”的风格。

Observable.fromArray("I", "AM", "CHINESE")
        .map(c -> c.substring(0, 1).toUpperCase() + c.substring(1).toLowerCase())
        .subscribe(c -> System.out.print(c + " "));

输出:

I Am Chinese 

参考:

http://www.vogella.com/tutorials/RxJava/article.html

http://www.cnblogs.com/aademeng/articles/7462540.html

https://www.jianshu.com/c/299d0a51fdd4

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏向治洪

android使用LruCache对listview加载图片时候优化处理

注意:LruCache是有版本限制的,低版本的sdk需要在libs文件夹添加相应的support-4v文件。 本文改造的大部分是参考http://www.i...

31060
来自专栏酷玩时刻

Retrofit2+Rxjava+MVP 实践

此博文根据前面两篇文章 Android MVP 架构初试 Android MVP 架构封装 再结合主流框架Retrofit2+Rxjava来个实践(实现聚合网...

43440
来自专栏Android常用基础

简单好用的RecyclerView适配器

给大家推荐个好用RecyclerView适配器,也就是在百篮应用里用到的适配器。为了大家学习方便,这里简单的写下如何使用这个适配器,不做代码具体的分析,有兴趣的...

16330
来自专栏Android干货

关于Activity销毁,而绘制UI的子线程未销毁出现的问题

38160
来自专栏全栈之路

android 实现本地定时推送(兼容)

首先写几点感悟: - 做兼容真的很累很费劲~ - android 8.0 广播部分不再支持动态注册,所以应该用service来实现定时推送功能 - 无论是闹钟还...

79420
来自专栏蜉蝣禅修之道

Android之使用JavaMail发送邮件

21120
来自专栏Android先生

Android 使用代码实现一个选词(拖拽)填空题

在上一篇文章《Android 使用代码实现一个填空题》中,我们学习了如何实现一个填空题,今天继续接着上一篇文章的节奏,学习一下如何实现一个选词填空题,由于本文中...

12820
来自专栏Android机动车

RxJava从入门到不离不弃(五)——do操作符

对于do系列操作符理解比较容易,他相当于给Observable执行周期的关键节点添加回调。当Observable执行到这个阶段的时候,这些回调就会被触发。

26730
来自专栏码匠的流水账

聊聊spring cloud gateway的RedirectToGatewayFilter

本文主要研究下spring cloud gateway的RedirectToGatewayFilter

20210
来自专栏俞其荣的博客

Android Architecture Component之Lifecycle解析HeaderLifecyclePart 1Part 2Part 3Footer

终于到了最后的关头,Android Architecture Component 系列的最后一节内容。今天给大家带来的就是 Lifecycle 的解析。

19640

扫码关注云+社区

领取腾讯云代金券