RxJava2学习笔记(1)

菩提树下的杨过

发布于 2018-03-28 13:51:58

7190

发布于 2018-03-28 13:51:58

作为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 + " "));

输出：