RxJava基本概念（一）

背景

RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

在Github上RxJava是这么介绍自己的，作为JVM的一个扩展，用于使用Java VM的可观察序列编写异步和基于事件的程序的库。ok，关键词：扩展、异步，基于事件。

概念

在开始之前，要先知道下面几个单词的意思，因为很容易记混了。

名称	含义
Observable	可被观察者，也就是被观察者（公众号）
Observer	观察者（人）
subscribe	订阅（主动关注该公众号）

举个栗子，微信的公众号就像是一个被观察者，当某人关注了这个公众号，就可以接收到公众号发送的消息，但如果没有关注该公众号，那么一切故事都不会发生。

简单订阅过程分析

创建一个Observable对象，使用Observable.create即可。

Observable<String> myObservable = Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            e.onNext("谢谢关注！");
            e.onComplete();
        }
    });

myObservable发出消息后调用onComplete()表示事件队列完成，用myObserver来处理刚才收到的内容。

Observer<String> myObserver = new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.print("onSubscribe");
        }

        @Override
        public void onNext(String value) {
            System.out.print("onNext" + value);
        }

        @Override
        public void onError(Throwable e) {
            System.out.print("onError");
        }

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

打印结果如下：

09-11 15:23:40.635 27057-27057/? I/zxf: onSubscribe
</br>09-11 15:23:40.635 27057-27057/? I/zxf: onNext谢谢关注！
</br>09-11 15:23:40.635 27057-27057/? I/zxf: onComplete

扩展

创建一个被观察者对象还有一种更为简洁的写法，用just(T...)可以添加多个事件内容，

Observable<String> myObservableEasy =
            Observable.just("谢谢关注!");

如果观察者只关心onNext()方法的，则只需要创建一个Consumer对象就可以了

Consumer<String> myConsumer = new Consumer<String>(){
        @Override
        public void accept(String s) throws Exception {
            Log.i("zxf","onNext" + s);
        }
    };

观察者也可以用Subscriber来创建，在使用上和Observer基本是一样的。

Subscriber<String> subscriber = new Subscriber<String>() {
        @Override
        public void onSubscribe(Subscription s) {

        }

        @Override
        public void onNext(String s) {

        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onComplete() {

        }
    };

注意

• ObservableEmitter提供了onNext()、onError()、onComplete()方法，onNext()可以发送多个事件，onError()代表事件中断，onComplete()则是消息完成，onComplete()和onError()是互斥的，有且只能存在一个。
• 不要搞错了订阅关系，Observable订阅Observer，看起来好像是弄反了，其实不是，只是考虑了流式API的特点。