Rxjava2入门教程二：Observable与Observer响应式编程在Rxjava2中的典型实现

蔚落 2023-07-13 13:48 31阅读 0赞

在RxJava中，函数响应式编程具体表现为一个观察者（Observer）订阅一个可观察对象（Observable），通过创建可观察对象发射数据流，经过一系列操作符（Operators）加工处理和线程调度器（Scheduler）在不同线程间的转发，最后由观察者接受并做出响应的一个过程  
ObservableSource与Observer是RxJava2中最典型的一组观察者与可观察对象的组合，其他四组可以看做是这一组的改进版或者简化版。

Observable

抽象类Observable是接口ObservableSource下的一个抽象实现，我们可以通过Observable创建一个可观察对象发射数据流。

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    emitter.onNext("Hello World");
                    emitter.onComplete();
                }
            });

调用Observable.create方法，创建一个可观察对象，并通过onNext发送一条数据“Hello World”，然后通过onComplete发送完成通知。

Observer

创建一个观察者Observer来接受并响应可观察对象发射的数据

Observer<String> observer = new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
    
                }
    
                @Override
                public void onNext(String s) {
                    System.out.println(s);
                }
    
                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                }
    
                @Override
                public void onComplete() {
                    System.out.println("接受完成");
                }
            };

在onNext方法中接收到可观察对象发射的数据"Hello World",并做出响应——打印到控制台。

Observer订阅Observable

observable.subscribe(observer);

通过subscribe方法，使Observer与Observable建立订阅关系，Observer与Observable便成为了一个整体，Observer便可对Observable中的行为作出响应。

Emitter/Observer

通过Observable.create创建可观察对象时，我们可以发现具体执行发射动作的是由接口ObservableEmitter的实例化对象完成的，而ObservableEmitter<T> 继承自 接口Emitter<T>，查看源码接口Emitter的具体代码如下：

public interface Emitter<T> {
            //用来发送数据，可多次调用，每调用一次发送一条数据
        void onNext(@NonNull T value);
            //用来发送异常通知，只发送一次，若多次调用只发送第一条
        void onError(@NonNull Throwable error);
            //用来发送完成通知，只发送一次，若多次调用只发送第一条
        void onComplete();
    }

onNext：用来发送数据，可多次调用，每调用一次发送一条数据  
onError：用来发送异常通知，只发送一次，若多次调用只发送第一条  
onComplete：用来发送完成通知，只发送一次，若多次调用只发送第一条

onError与onComplete互斥，两个方法只能调用一个不能同时调用

数据在发送时，出现异常可以调用onError发送异常通知也可以不调用，因为其所在的方法subscribe会抛出异常，

若数据在全部发送完之后均正常，可以调用onComplete发送一条完成通知

接口Observer中的三个方法（onNext,onError,onComplete）正好与Emitter中的三个方法相对应，对于Emitter中对应方法发送的数据或通知进行响应。

步骤简化

去掉中间变量可以对之前的代码简化为以下形式：

public void demo2() {
            Observable
                    .create(new ObservableOnSubscribe<String>() {
                        @Override
                        public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                            emitter.onNext("Hello World");
                            emitter.onComplete();
                        }
                    })
                    .subscribe(new Observer<String>() {
                        @Override
                        public void onSubscribe(Disposable d) {
                        }
    
                        @Override
                        public void onNext(String s) {
                            System.out.println(s);
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            e.printStackTrace();
                        }
    
                        @Override
                        public void onComplete() {
                            System.out.println("接受完成");
                        }
                    });
        }

再应用Rxjava中强大的操作符，可以将代码简化成以下形式：

public void demo3() {
            Observable.just("Hello World")
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(@NonNull String s) throws Exception {
                            System.out.println(s);
                        }
                    });
        }

再通过λ表达式，可进一步简化

public void demo3_1() {
            Observable.just("Hello World").subscribe(System.out::println);
        }

其中，just操作符可用来发送单条数据，数字，字符串，数组，对象，集合都可以当做单条数据发送。  
Consumer可以看做是对观察者Observer功能单一化之后的产物——消费者，上例中的Consumer通过其函数accept只接收可观察对象发射的数据，不接收异常信息或完成信息。  
如果想接收异常信息或完成信息可以用下面的代码：

public void demo4() {
            Observable.just("Hello World")
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(@NonNull String s) throws Exception {
                            System.out.println(s);
                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(@NonNull Throwable throwable) throws Exception {
                            throwable.printStackTrace();
                        }
                    }, new Action() {
                        @Override
                        public void run() throws Exception {
                            System.out.println("接受完成");
                        }
                    });
        }

第二个参数Consumer规定泛型<Throwable>通过函数accept接收异常信息。  
第三个参数Action也是对观察者Observer功能单一化之后的产物--行动，通过函数run接收完成信息，作出响应行动。

发送数据序列

Observable可以发送单条数据也可以发送数据序列  
通过最基础的方法发送数据序列：

public void demo5(final List<String> list) {
            Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    try {
                        for (String str : list) {
                            emitter.onNext(str);
                        }
                        emitter.onComplete();
                    } catch (Exception e) {
                        emitter.onError(e);
                    }
                }
            }).subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(Disposable d) {
                }
    
                @Override
                public void onNext(String s) {
                    System.out.println(s);
                }
    
                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                }
    
                @Override
                public void onComplete() {
                    System.out.println("接受完成");
                }
            });
        }

在subscribe方法中，遍历集合list中的String元素，通过emitter.onNext(str)逐一发送；发送完成后通过emitter.onComplete()发送完成通知；如果发送过程中遇到异常，通过emitter.onError(e)发送异常信息。  
Observer中通过onNext接收emitter发送的每一条信息并打印到控制台（emitter发送几次，Observer便接收几次），通过onError(Throwable e)接收异常信息，onComplete()接收完成信息。  
同样可以通过操作符对其进行简化，如下;

public void demo6(final List<String> list) {
            Observable
                    .fromIterable(list)
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(@NonNull String s) throws Exception {
                            System.out.println(s);
                        }
                    });
        }

再用λ表达式，进一步简化

public void demo6_1(final List<String> list) {
            Observable.fromIterable(list).subscribe(System.out::println);
        }

其中fromIterable操作符，可用来将一个可迭代对象中的元素逐一发送

Disposable

在之前的例子中，可以看到Observer接口中还有一个方法

public void onSubscribe(Disposable d) {
    }

是在观察者Observer与可观察对象Observable，建立订阅关系后，回调这个方法，并且传过来一个Disposable类型的参数，可通过Disposable来控制Observer与Observable之间的订阅。  
无论观察者Observer以何种方式订阅可观察对象Observable，都会生成一个Disposable，如下：

public void demo7(final List<String> list) {
            Disposable disposable1 = Observable.just("Hello World")
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(@NonNull String s) throws Exception {
                            System.out.println(s);
                        }
                    });
            Disposable disposable2 = Observable
                    .fromIterable(list)
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(@NonNull String s) throws Exception {
                            System.out.println(s);
                        }
                    });
        }

查看Disposable接口的源码，如下：

public interface Disposable {
            void dispose();
            boolean isDisposed();
    }

其中isDisposed()方法用来判断当前订阅是否失效，dispose()方法用来取消当前订阅。  
只有当观察者Observer与可观察对象Observable之间建立订阅关系，并且订阅关系有效时，Observer才能对Observable进行响应。如果Observer在响应Observable的过程中，订阅关系被取消，则Observer无法对取消订阅关系之后Observable的行为作出响应。  
运行下面的代码，当Observable接收到第5条数据时，取消订阅关系。

public void demo8() {
            Observable.create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    for (int i = 0; i < 10; i++) {
                        System.out.println("发送" + i);
                        emitter.onNext(i);
                    }
                    emitter.onComplete();
                }
            }).subscribe(new Observer<Integer>() {
                private Disposable disposable;
    
                @Override
                public void onSubscribe(Disposable d) {
                    disposable = d;
                }
    
                @Override
                public void onNext(Integer integer) {
                    System.out.println("接收" + integer);
                    if (integer > 4) disposable.dispose();
                }
    
                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                }
    
                @Override
                public void onComplete() {
                    System.out.println("数据接受完成");
                }
            });
        }

控制台日志如下：

I/System.out: 发送0
     I/System.out: 接收0
     I/System.out: 发送1
     I/System.out: 接收1
     I/System.out: 发送2
     I/System.out: 接收2
     I/System.out: 发送3
     I/System.out: 接收3
     I/System.out: 发送4
     I/System.out: 接收4
     I/System.out: 发送5
     I/System.out: 接收5
     I/System.out: 发送6
     I/System.out: 发送7
     I/System.out: 发送8
     I/System.out: 发送9

可以发现取消订阅关系之前，Observable发送一条数据，Observe便接收一条，但是取消订阅关系之后，Observe将不再接收Observable发送的数据。