在面向对象的架构中,开发者致力于创建一组解耦的实体。这样的话,实体就可以在不用妨碍整个系统的情况下可以被测试、复用和维护。设计这种系统就带来一个棘手的负面影响:维护相关对象之间的统一。
在Smalltalk MVC架构中,创建模式的第一个例子就是用来解决这个问题的。用户界面框架提供一种途径使UI元素与包含数据的实体对象相分离,并且同时,它提供一种灵活的方法来保持它们之间的同步。
在这本畅销的四人组编写的《设计模式——可复用面向对象软件的基础》一书中,观察者模式是最有名的设计模式之一。它是一种行为模式并提供一种以一对多的依赖来绑定对象的方法:即当一个对象发生变化时,依赖它的所有对象都会被通知并且会自动更新。
在本章中,我们将会对观察者模式有一个概述,它是如何实现的以及如何用RxJava来扩展,Observable是什么,以及Observables如何与Iterables相关联。
在今天,观察者模式是出现的最常用的软件设计模式之一。它基于subject这个概念。subject是一种特殊对象,当它改变时,那些由它保存的一系列对象将会得到通知。而这一系列对象被称作Observers,它们会对外暴漏了一个通知方法,当subject状态发生变化时会调用的这个方法。
在上一章中,我们看到了电子表单的例子。现在我们可以展开这个例子讲,展示一个更复杂的场景。让我们考虑这样一个填着账户数据的电子表单。我们可以把这些数据比作一张表,或者是3D柱状图,或者是饼状图。它们中每一个代表的意义都取决于同一组要展示的数据。每一个都是一个观察者,都依赖于那一个subject,维护着全部信息。
3D柱状图这个类、饼状图类、表这个类以及维护这些数据的类是完全解耦的:它们彼此相互独立复用,但也能协同工作。这些表示类彼此不清楚对方,但是正如它们所做的:它们知道在哪能找到它们需要展示的信息,它们也知道一旦数据发生变化就通知需要更新数据表示的那个类。
这有一张图描述了Subject/Observer的关系是怎样的一对多的关系:
上面这张图展示了一个Subject为3个Observers提供服务。很明显,没有理由去限制Observers的数量:如果有需要,一个Subject可以有无限多个Observers,当subject状态发生变化时,这些Observers中的每一个都会收到通知。
观察者模式很适合下面这些场景中的任何一个:
在RxJava的世界里,我们有四种角色:
Observables和Subjects是两个“生产”实体,Observers和Subscribers是两个“消费”实体。
当我们异步执行一些复杂的事情,Java提供了传统的类,例如Thread、Future、FutureTask、CompletableFuture来处理这些问题。当复杂度提升,这些方案就会变得麻烦和难以维护。最糟糕的是,它们都不支持链式调用。
RxJava Observables被设计用来解决这些问题。它们灵活,且易于使用,也可以链式调用,并且可以作用于单个结果程序上,更有甚者,也可以作用于序列上。无论何时你想发射单个标量值,或者一连串值,甚至是无穷个数值流,你都可以使用Observable。
Observable的生命周期包含了三种可能的易于与Iterable生命周期事件相比较的事件,下表展示了如何将Observable async/push 与 Iterable sync/pull相关联起来。
Event | Iterable(pull) | Observable(push) |
---|---|---|
检索数据 | T next() |
onNext(T) |
发现错误 | throws Exception |
onError(Throwable) |
完成 | !hasNext() |
onCompleted() |
使用Iterable时,消费者从生产者那里以同步的方式得到值,在这些值得到之前线程处于阻塞状态。相反,使用Observable时,生产者以异步的方式把值推给观察者,无论何时,这些值都是可用的。这种方法之所以更灵活是因为即便值是同步或异步方式到达,消费者在这两种场景都可以根据自己的需要来处理。
为了更好地复用Iterable接口,RxJava Observable类扩展了GOF观察者模式的语义。引入了两个新的接口:
从发射物的角度来看,有两种不同的Observables:热的和冷的。一个"热"的Observable典型的只要一创建完就开始发射数据,因此所有后续订阅它的观察者可能从序列中间的某个位置开始接受数据(有一些数据错过了)。一个"冷"的Observable会一直等待,直到有观察者订阅它才开始发射数据,因此这个观察者可以确保会收到整个数据序列。
在接下来的小节中将讨论Observables提供的两种创建Observable的方法。
create()方法使开发者有能力从头开始创建一个Observable。它需要一个OnSubscribe对象,这个对象继承Action1,当观察者订阅我们的Observable时,它作为一个参数传入并执行call()函数。
Observable.create(new Observable.OnSubscribe<Object>(){
@Override
public void call(Subscriber<? super Object> subscriber) {
}
});
Observable通过使用subscriber变量并根据条件调用它的方法来和观察者通信。让我们看一个“现实世界”的例子:
Observable<Integer> observableString = Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> observer) {
for (int i = 0; i < 5; i++) {
observer.onNext(i);
}
observer.onCompleted();
}
});
Subscription subscriptionPrint = observableString.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("Observable completed");
}
@Override
public void onError(Throwable e) {
System.out.println("Oh,no! Something wrong happened!");
}
@Override
public void onNext(Integer item) {
System.out.println("Item is " + item);
}
});
例子故意写的简单,是因为即便是你第一次见到RxJava的操作,我想让你明白接下来要发生什么。
我们创建一个新的Observable<Integer>
,它执行了5个元素的for循环,一个接一个的发射他们,最后完成。
另一方面,我们订阅了Observable,返回一个Subscription 。一旦我们订阅了,我们就开始接受整数,并一个接一个的打印出它们。我们并不知道要接受多少整数。事实上,我们也无需知道是因为我们为每种场景都提供对应的处理操作:
在上一个例子中,我们创建了一个整数序列并一个一个的发射它们。假如我们已经有一个列表呢?我们是不是可以不用for循环而也可以一个接一个的发射它们呢?
在下面的例子代码中,我们从一个已有的列表中创建一个Observable序列:
List<Integer> items = new ArrayList<Integer>();
items.add(1);
items.add(10);
items.add(100);
items.add(200);
Observable<Integer> observableString = Observable.from(items);
Subscription subscriptionPrint = observableString.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("Observable completed");
}
@Override
public void onError(Throwable e) {
System.out.println("Oh,no! Something wrong happened!");
}
@Override
public void onNext(Integer item) {
System.out.println("Item is " + item);
}
});
输出的结果和上面的例子绝对是一样的。
from()
创建符可以从一个列表/数组来创建Observable,并一个接一个的从列表/数组中发射出来每一个对象,或者也可以从Java Future
类来创建Observable,并发射Future对象的.get()
方法返回的结果值。传入Future
作为参数时,我们可以指定一个超时的值。Observable将等待来自Future
的结果;如果在超时之前仍然没有结果返回,Observable将会触发onError()
方法通知观察者有错误发生了。
如果我们已经有了一个传统的Java函数,我们想把它转变为一个Observable又改怎么办呢?我们可以用create()
方法,正如我们先前看到的,或者我们也可以像下面那样使用以此来省去许多模板代码:
Observable<String> observableString = Observable.just(helloWorld());
Subscription subscriptionPrint = observableString.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
System.out.println("Observable completed");
}
@Override
public void onError(Throwable e) {
System.out.println("Oh,no! Something wrong happened!");
}
@Override
public void onNext(String message) {
System.out.println(message);
}
});
helloWorld()
方法比较简单,像这样:
private String helloWorld(){
return "Hello World";
}
不管怎样,它可以是我们想要的任何函数。在刚才的例子中,我们一旦创建了Observable,just()
执行函数,当我们订阅Observable时,它就会发射出返回的值。
just()
方法可以传入一到九个参数,它们会按照传入的参数的顺序来发射它们。just()
方法也可以接受列表或数组,就像from()
方法,但是它不会迭代列表发射每个值,它将会发射整个列表。通常,当我们想发射一组已经定义好的值时会用到它。但是如果我们的函数不是时变性的,我们可以用just来创建一个更有组织性和可测性的代码库。
最后注意just()
创建符,它发射出值后,Observable正常结束,在上面那个例子中,我们会在控制台打印出两条信息:“Hello World”和“Observable completed”。
当我们需要一个Observable毫无理由的不再发射数据正常结束时,我们可以使用empty()
。我们可以使用never()
创建一个不发射数据并且也永远不会结束的Observable。我们也可以使用throw()
创建一个不发射数据并且以错误结束的Observable。
subject
是一个神奇的对象,它可以是一个Observable同时也可以是一个Observer:它作为连接这两个世界的一座桥梁。一个Subject可以订阅一个Observable,就像一个观察者,并且它可以发射新的数据,或者传递它接受到的数据,就像一个Observable。很明显,作为一个Observable,观察者们或者其它Subject都可以订阅它。
一旦Subject订阅了Observable,它将会触发Observable开始发射。如果原始的Observable是“冷”的,这将会对订阅一个“热”的Observable变量产生影响。
RxJava提供四种不同的Subject:
Publish是Subject的一个基础子类。让我们看看用PublishSubject实现传统的Observable Hello World
:
PublishSubject<String> stringPublishSubject = PublishSubject.create();
Subscription subscriptionPrint = stringPublishSubject.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
System.out.println("Observable completed");
}
@Override
public void onError(Throwable e) {
System.out.println("Oh,no!Something wrong happened!");
}
@Override
public void onNext(String message) {
System.out.println(message);
}
});
stringPublishSubject.onNext("Hello World");
在刚才的例子中,我们创建了一个PublishSubject
,用create()
方法发射一个String
值,然后我们订阅了PublishSubject。此时,没有数据要发送,因此我们的观察者只能等待,没有阻塞线程,也没有消耗资源。就在这随时准备从subject接收值,如果subject没有发射值那么我们的观察者就会一直在等待。再次声明的是,无需担心:观察者知道在每个场景中该做什么,我们不用担心什么时候是因为它是响应式的:系统会响应。我们并不关心它什么时候响应。我们只关心它响应时该做什么。
最后一行代码展示了手动发射字符串“Hello World”,它触发了观察者的onNext()
方法,让我们在控制台打印出“Hello World”信息。
让我们看一个更复杂的例子。话说我们有一个private
声明的Observable,外部不能访问。Observable在它生命周期内发射值,我们不用关心这些值,我们只关心他们的结束。
首先,我们创建一个新的PublishSubject来响应它的onNext()
方法,并且外部也可以访问它。
final PublishSubject<Boolean> subject = PublishSubject.create();
subject.subscribe(new Observer<Boolean>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Boolean aBoolean) {
System.out.println("Observable Completed");
}
});
然后,我们创建“私有”的Observable,只有subject才可以访问的到。
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();
}
}).doOnCompleted(new Action0() {
@Override
public void call() {
subject.onNext(true);
}
}).subscribe();
Observable.create()
方法包含了我们熟悉的for循环,发射数字。doOnCompleted()
方法指定当Observable结束时要做什么事情:在subject上发射true。最后,我们订阅了Observable。很明显,空的subscribe()
调用仅仅是为了开启Observable,而不用管已发出的任何值,也不用管完成事件或者错误事件。为了这个例子我们需要它像这样。
在这个例子中,我们创建了一个可以连接Observables并且同时可被观测的实体。当我们想为公共资源创建独立、抽象或更易观测的点时,这是极其有用的。
简单的说,BehaviorSubject会首先向他的订阅者发送截至订阅前最新的一个数据对象(或初始值),然后正常发送订阅后的数据流。
BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create(1);
在这个短例子中,我们创建了一个能发射整形(Integer)的BehaviorSubject。由于每当Observes订阅它时就会发射最新的数据,所以它需要一个初始值。
ReplaySubject会缓存它所订阅的所有数据,向任意一个订阅它的观察者重发:
ReplaySubject<Integer> replaySubject = ReplaySubject.create();
当Observable完成时AsyncSubject只会发布最后一个数据给已经订阅的每一个观察者。
AsyncSubject<Integer> asyncSubject = AsyncSubject.create();
本章中,我们了解到了什么是观察者模式,为什么Observables在今天的编程场景中如此重要,以及如何创建Observables和subjects。
下一章中,我们将创建第一个基于RxJava的Android应用程序,学习如何检索数据来填充listview,以及探索如何创建一个基于RxJava的响应式UI。
京东创始人刘强东和其妻子章泽天最近成为了互联网舆论关注的焦点。有关他们“移民美国”和在美国购买豪宅的传言在互联网上广泛传播。然而,京东官方通过微博发言人发布的消息澄清了这些传言,称这些言论纯属虚假信息和蓄意捏造。
日前,据博主“@超能数码君老周”爆料,国内三大运营商中国移动、中国电信和中国联通预计将集体采购百万台规模的华为Mate60系列手机。
据报道,荷兰半导体设备公司ASML正看到美国对华遏制政策的负面影响。阿斯麦(ASML)CEO彼得·温宁克在一档电视节目中分享了他对中国大陆问题以及该公司面临的出口管制和保护主义的看法。彼得曾在多个场合表达了他对出口管制以及中荷经济关系的担忧。
今年早些时候,抖音悄然上线了一款名为“青桃”的 App,Slogan 为“看见你的热爱”,根据应用介绍可知,“青桃”是一个属于年轻人的兴趣知识视频平台,由抖音官方出品的中长视频关联版本,整体风格有些类似B站。
日前,威马汽车首席数据官梅松林转发了一份“世界各国地区拥车率排行榜”,同时,他发文表示:中国汽车普及率低于非洲国家尼日利亚,每百户家庭仅17户有车。意大利世界排名第一,每十户中九户有车。
近日,一项新的研究发现,维生素 C 和 E 等抗氧化剂会激活一种机制,刺激癌症肿瘤中新血管的生长,帮助它们生长和扩散。
据媒体援引消息人士报道,苹果公司正在测试使用3D打印技术来生产其智能手表的钢质底盘。消息传出后,3D系统一度大涨超10%,不过截至周三收盘,该股涨幅回落至2%以内。
9月2日,坐拥千万粉丝的网红主播“秀才”账号被封禁,在社交媒体平台上引发热议。平台相关负责人表示,“秀才”账号违反平台相关规定,已封禁。据知情人士透露,秀才近期被举报存在违法行为,这可能是他被封禁的部分原因。据悉,“秀才”年龄39岁,是安徽省亳州市蒙城县人,抖音网红,粉丝数量超1200万。他曾被称为“中老年...
9月3日消息,亚马逊的一些股东,包括持有该公司股票的一家养老基金,日前对亚马逊、其创始人贝索斯和其董事会提起诉讼,指控他们在为 Project Kuiper 卫星星座项目购买发射服务时“违反了信义义务”。
据消息,为推广自家应用,苹果现推出了一个名为“Apps by Apple”的网站,展示了苹果为旗下产品(如 iPhone、iPad、Apple Watch、Mac 和 Apple TV)开发的各种应用程序。
特斯拉本周在美国大幅下调Model S和X售价,引发了该公司一些最坚定支持者的不满。知名特斯拉多头、未来基金(Future Fund)管理合伙人加里·布莱克发帖称,降价是一种“短期麻醉剂”,会让潜在客户等待进一步降价。
据外媒9月2日报道,荷兰半导体设备制造商阿斯麦称,尽管荷兰政府颁布的半导体设备出口管制新规9月正式生效,但该公司已获得在2023年底以前向中国运送受限制芯片制造机器的许可。
近日,根据美国证券交易委员会的文件显示,苹果卫星服务提供商 Globalstar 近期向马斯克旗下的 SpaceX 支付 6400 万美元(约 4.65 亿元人民币)。用于在 2023-2025 年期间,发射卫星,进一步扩展苹果 iPhone 系列的 SOS 卫星服务。
据报道,马斯克旗下社交平台𝕏(推特)日前调整了隐私政策,允许 𝕏 使用用户发布的信息来训练其人工智能(AI)模型。新的隐私政策将于 9 月 29 日生效。新政策规定,𝕏可能会使用所收集到的平台信息和公开可用的信息,来帮助训练 𝕏 的机器学习或人工智能模型。
9月2日,荣耀CEO赵明在采访中谈及华为手机回归时表示,替老同事们高兴,觉得手机行业,由于华为的回归,让竞争充满了更多的可能性和更多的魅力,对行业来说也是件好事。
《自然》30日发表的一篇论文报道了一个名为Swift的人工智能(AI)系统,该系统驾驶无人机的能力可在真实世界中一对一冠军赛里战胜人类对手。
近日,非营利组织纽约真菌学会(NYMS)发出警告,表示亚马逊为代表的电商平台上,充斥着各种AI生成的蘑菇觅食科普书籍,其中存在诸多错误。
社交媒体平台𝕏(原推特)新隐私政策提到:“在您同意的情况下,我们可能出于安全、安保和身份识别目的收集和使用您的生物识别信息。”
2023年德国柏林消费电子展上,各大企业都带来了最新的理念和产品,而高端化、本土化的中国产品正在不断吸引欧洲等国际市场的目光。
罗永浩日前在直播中吐槽苹果即将推出的 iPhone 新品,具体内容为:“以我对我‘子公司’的了解,我认为 iPhone 15 跟 iPhone 14 不会有什么区别的,除了序(列)号变了,这个‘不要脸’的东西,这个‘臭厨子’。