RxJava操作符实践:6_辅助操作之6_subscribe

一、描述

操作来自Observable的发射物和通知。

Subscribe操作符是连接观察者和Observable的胶水。一个观察者要想看到Observable发射的数据项,或者想要从Observable获取错误和完成通知,它首先必须使用这个操作符订阅那个Observable。

Subscribe操作符的一般实现可能会接受一到三个方法(然后由观察者组合它们),或者接受一个实现了包含这三个方法的接口的对象(有时叫做Observer或Subscriber):

onNext

每当Observable发射了一项数据它就会调用这个方法。这个方法的参数是这个Observable发射的数据项。

onError

Observable调用这个方法表示它无法生成期待的数据或者遇到了其它错误。这将停止Observable,它在这之后不会再调用onNext或onCompleted。onError方法的参数是导致这个错误的原因的一个表示(有时可能是一个Exception或Throwable对象,其它时候也可能是一个简单的字符串,取决于具体的实现)。

onCompleted

如果没有遇到任何错误,Observable在最后一次调用onCompleted之后会调用这个方法。

如果一个Observable直到有一个观察者订阅它才开始发射数据项,就称之为”冷”的Observable;如果一个Observable可能在任何时刻开始发射数据,就称之为”热”的Observable,一个订阅者可能从开始之后的某个时刻开始观察它发射的数据序列,它可能会错过在订阅之前发射的数据。

RxJava中的实现是subscribe方法。

如果你使用无参数的版本,它将触发对Observable的一个订阅,但是将忽略它的发射物和通知。这个操作会激活一个”冷”的Observable。

你也可以传递一到三个函数给它,它们会按下面的方法解释:

  1. onNext
  2. onNext和onError
  3. onNext, onError和onCompleted

最后,你还可以传递一个Observer或Subscriber接口给它,Observer接口包含这三个以on开头的方法。Subscriber接口也实现了这三个方法,而且还添加了几个额外的方法,用于支持使用反压操作(reactive pull backpressure),这让Subscriber可以在Observable完成前取消订阅。

subscribe方法返回一个实现了Subscription接口的对象。这个接口包含unsubscribe方法,任何时刻你都可以调用它来断开subscribe方法建立的Observable和观察者之间的订阅关系。

二、示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
Observable.just(1, 2)
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
System.out.println("onCompleted.");
}

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

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

三、运行结果

1
2
3
onNext: 1
onNext: 2
onCompleted.

四、参考资料

ReactiveX官方文档

ReactiveX文档中文翻译

PS:欢迎关注 SherlockShi 个人博客

感谢你的支持,让我继续努力分享有用的技术和知识点!