RxJava学习笔记（基本用法）

首先这是基于rxjava1.0的学习，最近又出了2.0版本，有了一些改动，首先用法就有了一定的变化，就比如开始我在Android Studio添加的依赖库选了2.0，结果测试rxjava-essentials中的代码时，发现方法根本找不到了。</br>
不过正所谓学习嘛，虽然说更新了2.0版本，但是内部的核心肯定是换汤不换药的，就从1.0版本开始学！

RxJava的与众不同

从纯Java的观点看，Rxjava Observale类源自于经典的Gang of Four的观察者模式。
它添加了三个缺少的功能：</br>

• 生产者在没有更多数据可用时能够发出信号通知：onCompleted()事件。
• 生产者在发生错误时能够发出信号通知：onError()事件。
• RxJava Observables 能够组合而不是嵌套，从而避免开发者陷入回调地狱。

在RxJava的世界里，我们有四种角色：

• Observable
• Observer
• Subscriber
• Subjects

Observables和Subjects是两个“生产”实体，Observers和Subscribers是两个“消
费”实体。

以上摘自《rxjava-essentials》

Observer是观察者接口，而Subscriber是其抽象实现类。

Observable.create()

• Observable.create()创建一个Observable（）；

• Observable.subscribe(Observer ob)订阅一个观察者。

    public static void main(String... args) {
        Observable.create(new Observable.OnSubscribe<Integer>() {
  
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 5; i++) {
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        
        }).subscribe(new Observer<Integer>() {
  
            @Override
            public void onCompleted() {
                System.out.println("Completed");
            }
  
            @Override
            public void onError(Throwable e) {
                System.out.println("Error");
            }
  
            @Override
            public void onNext(Integer integer) {
                System.out.println("i = " + integer);
            }
        });
  
    }
  

打印结果：</br>
i = 0</br>
i = 1</br>
i = 2</br>
i = 3</br>
i = 4</br>
Completed</br>

除了可以用Observable.create(),同时还有Observable.from(),Observable.just()。

Observable.from()

• Observable.from()可以直接将我们的列表丢进参数中。

    public static void main(String... args) {
    List<Integer> items = new ArrayList<>();
    items.add(1);
    items.add(11);
    items.add(111);
    items.add(1111);
    
    Observable.from(items)
            .subscribe(new Observer<Integer>() {
    
                @Override
                public void onCompleted() {
                    System.out.println("Completed");
                }
    
                @Override
                public void onError(Throwable e) {
                    System.out.println("Error");
                }
    
                @Override
                public void onNext(Integer integer) {
                    System.out.println("i = " + integer);
                }
            });
    }
  

打印结果：</br>
i = 0</br>
i = 1</br>
i = 2</br>
i = 3</br>
i = 4</br>
Completed</br>

Observable.just()

• Observable.just()可以将我们现有的方法转换成Observable。

    public static void main(String... args) {
        Observable.just(reInt())
                .subscribe(new Observer<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Completed");
            }
  
            @Override
            public void onError(Throwable e) {
                System.out.println("Error");
            }
  
            @Override
            public void onNext(Integer integer) {
                System.out.println("i = " + integer);
            }
        });
    }
  
    public static int reInt(){
        return 12;
    }
  

打印结果：</br>
i = 12</br>
Completed</br>

Subject

• 前面对Observer,Observable,Subscriber都有用到，单单还有一个Subject。
• Subject = Observable + Observer,它可以是一个Observable同时也可以是一个Observer，它作为连接这两个世界的一座桥梁。
• 一个Subject可以订阅一个Observable，就像一个观察者，并且它可以发射新的数据，或者传递它接受到的数据，就像一个Observable。很明显，作为一个Observable，观察者们或者其它Subject都可以订阅它。

Rxjava提供了四种不同的Subject:

• PublishSubject
• BehaviorSubject
• ReplaySubject
• AsyncSubject

PublishSubject

public static void main(String... args) {
    PublishSubject<String> publishSubject = PublishSubject.create();
    publishSubject.subscribe(new Observer<Object>() {
        @Override
        public void onCompleted() {
            System.out.println("Completed");
        }

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

        @Override
        public void onNext(Object o) {
            System.out.println(o.toString());
        }
    });
    publishSubject.onNext("HelloWorld");
}

打印结果：</br>
HelloWorld

• 代码开始创建了一个PublishSubject，用create()发射一个String值，之后订阅了PublishSubject。一直到这里，也没有数据要发送，直到最后publishSubject.onNext("HelloWorld")触发了观察者的onNext()方法，将传递过来的"HelloWorld"打印出来。

BehaviorSubject

• BehaviorSubject会首先向他的订阅者发送截至订阅前最新的一个数据对象或者初始值，然后正常的发送订阅后的数据流。

    public static void main(String... args) {
        BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create(1);
        behaviorSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onCompleted() {
  
            }
  
            @Override
            public void onError(Throwable e) {
  
            }
  
            @Override
            public void onNext(Integer integer) {
                System.out.println("i = " + integer);
            }
        });
        behaviorSubject.onNext(999);
    }
  

打印结果：</br>
i = 1</br>
i = 999</br>
首先接受到创建Observer时的初始值1，然后出发观察者的onNext()方法。

ReplaySubject

ReplaySubject会缓存它所订阅的所有数据,向任意一个订阅它的观察者重发。

public static void main(String... args) {
    ReplaySubject<Integer> replaySubject = ReplaySubject.create();
    replaySubject.onNext(333);
    replaySubject.onNext(444);
    replaySubject.onNext(555);
    replaySubject.onCompleted();

    replaySubject.subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("i = " + integer);
        }
    });
}

可以看到，replaySubject先发出了信息，然后才订阅了Observer。</br>
打印结果：</br>
i = 333</br>
i = 444</br>
i = 555</br>

AsyncSubject

当Observable完成时AsyncSubject只会发布最后一个数据给已经订阅的每一个观察者。

public static void main(String... args) {
    AsyncSubject<Integer> asyncSubject = AsyncSubject.create();
    asyncSubject.subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("i = " + integer);
        }
    });
    asyncSubject.onNext(333);
    asyncSubject.onNext(444);
    asyncSubject.onNext(555);
    asyncSubject.onCompleted();
}

在这里，asyncSubject发出了三个信息，但是根据asyncSubject的特性，只会发布最后一个数据给订阅的观察者，注意最后这一行代码：asyncSubject.onCompleted()，如果不加这一行代码，程序不能识别555是最后一个数据，将不会打印任何结果。</br>
打印结果：</br>
i = 555</br>