共计 8566 个字符，预计需要花费 22 分钟才能阅读完成。
多个订阅者的默认行为并不总是可取的。在本文中，我们将介绍如何更改此行为并以适当的方式处理多个订阅者。
但首先，让我们来看看多个订阅者的默认行为。
默认行为
假设我们有以下 Observable：
private static Observable getObservable() {
return Observable.create(subscriber -> {
subscriber.onNext(gettingValue(1));
subscriber.onNext(gettingValue(2));
subscriber.add(Subscriptions.create(() -> {
LOGGER.info(“Clear resources”);
}));
});
}
订阅者订阅后会立即发出两个元素。
在我们的示例中，我们有两个订阅者：
LOGGER.info(“Subscribing”);
Subscription s1 = obs.subscribe(i -> LOGGER.info(“subscriber#1 is printing ” + i));
Subscription s2 = obs.subscribe(i -> LOGGER.info(“subscriber#2 is printing ” + i));
s1.unsubscribe();
s2.unsubscribe();
想象一下，获取每个元素是一项代价高昂的操作 – 例如，它可能包括密集计算或打开 URL 连接。
为了简单起见，我们只返回一个数字：
private static Integer gettingValue(int i) {
LOGGER.info(“Getting ” + i);
return i;
}
这是输出：
Subscribing
Getting 1
subscriber#1 is printing 1
Getting 2
subscriber#1 is printing 2
Getting 1
subscriber#2 is printing 1
Getting 2
subscriber#2 is printing 2
Clear resources
Clear resources
我们可以看到，在默认情况下，获取每个元素和清除资源都要执行两次 - 对于每个订阅服务器一次。这不是我们想要的。ConnectableObservable 类有助于解决这个问题。
ConnectableObservable
ConnectableObservable 类允许与多个订阅者共享订阅，而不允许多次执行底层操作。
但首先，让我们创建一个 ConnectableObservable。
publish()
publish() 方法是从 Observable 创建一个 ConnectableObservable：
ConnectableObservable obs = Observable.create(subscriber -> {
subscriber.onNext(gettingValue(1));
subscriber.onNext(gettingValue(2));
subscriber.add(Subscriptions.create(() -> {
LOGGER.info(“Clear resources”);
}));
}).publish();
但就目前而言，它什么都不做。它的工作原理是 connect() 方法。
connect()
在调用 ConnectableObservable 的 connect() 方法之前，即使有一些订阅者，也不会触发 Observable 的 onSubcribe() 回调。
让我们来证明一下：
LOGGER.info(“Subscribing”);
obs.subscribe(i -> LOGGER.info(“subscriber #1 is printing ” + i));
obs.subscribe(i -> LOGGER.info(“subscriber #2 is printing ” + i));
Thread.sleep(1000);
LOGGER.info(“Connecting”);
Subscription s = obs.connect();
s.unsubscribe();
我们订阅，然后等待一秒钟再连接输出是:
Subscribing
Connecting
Getting 1
subscriber #1 is printing 1
subscriber #2 is printing 1
Getting 2
subscriber #1 is printing 2
subscriber #2 is printing 2
Clear resources
我们可以看到：
获取元素只出现一次我们想要的
清算资源也只出现一次
订阅后获取元素开始一秒钟
订阅不再触发元素的发射。只有 connect() 才能这样做
这种延迟可能是有益的 – 有时我们需要为所有订阅者提供相同的元素序列，即使其中一个订阅者比另一个订阅者更早。
可观察的一致视图 – 在 subscribe() 之后的 connect()
这个用例无法在我们之前的 Observable 上进行演示，因为它运行很冷，而且两个订阅者都可以获得整个元素序列。
相反，想象一下，元素发射不依赖于订阅的时刻，例如，鼠标点击发出的事件。现在还想象第二个订阅者在第一个订阅者之后订阅第二个订阅者。
第一个订阅者将获得此示例中发出的所有元素，而第二个订阅者将只接收一些元素。
另一方面，在正确的位置使用 connect() 方法可以为两个订阅者提供 Observable 序列上的相同视图。
让我们创建一个 Observable。它将在 JFrame 上点击鼠标时发出元素。
每个元素都是点击的 x 坐标：
private static Observable getObservable() {
return Observable.create(subscriber -> {
frame.addMouseListener(new MouseAdapter() {
@Override
public void mouseClicked(MouseEvent e) {
subscriber.onNext(e.getX());
}
});
subscriber.add(Subscriptions.create(() {
LOGGER.info(“Clear resources”);
for (MouseListener listener : frame.getListeners(MouseListener.class)) {
frame.removeMouseListener(listener);
}
}));
});
}
现在，如果我们以第二个间隔一个接一个地订阅两个订阅者，运行程序并开始单击，我们将看到第一个订阅者将获得更多元素：
public static void defaultBehaviour() throws InterruptedException {
Observable obs = getObservable();
LOGGER.info(“subscribing #1”);
Subscription subscription1 = obs.subscribe((i) ->
LOGGER.info(“subscriber#1 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“subscribing #2”);
Subscription subscription2 = obs.subscribe((i) ->
LOGGER.info(“subscriber#2 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“unsubscribe#1”);
subscription1.unsubscribe();
Thread.sleep(1000);
LOGGER.info(“unsubscribe#2”);
subscription2.unsubscribe();
}
subscribing #1
subscriber#1 is printing x-coordinate 280
subscriber#1 is printing x-coordinate 242
subscribing #2
subscriber#1 is printing x-coordinate 343
subscriber#2 is printing x-coordinate 343
unsubscribe#1
clearing resources
unsubscribe#2
clearing resources
connect() After subscribe()
为了使两个订阅者获得相同的序列，我们将 Observable 转换为 ConnectableObservable 并在订阅者之后调用 connect()：
public static void subscribeBeforeConnect() throws InterruptedException {
ConnectableObservable obs = getObservable().publish();
LOGGER.info(“subscribing #1”);
Subscription subscription1 = obs.subscribe(
i -> LOGGER.info(“subscriber#1 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“subscribing #2”);
Subscription subscription2 = obs.subscribe(
i -> LOGGER.info(“subscriber#2 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“connecting:”);
Subscription s = obs.connect();
Thread.sleep(1000);
LOGGER.info(“unsubscribe connected”);
s.unsubscribe();
}
现在他们将得到相同的序列：
subscribing #1
subscribing #2
connecting:
subscriber#1 is printing x-coordinate 317
subscriber#2 is printing x-coordinate 317
subscriber#1 is printing x-coordinate 364
subscriber#2 is printing x-coordinate 364
unsubscribe connected
clearing resources
所以重点是等待所有用户准备就绪然后调用 connect()。
在 Spring 应用程序中，我们可以在应用程序启动期间订阅所有组件，例如在 onApplicationEvent() 中调用 connect()。
让我们回到我们的例子; 注意，connect() 方法之前的所有单击操作都失败了。如果我们不想遗漏元素，但相反，我们可以在代码中更早地放置 connect()，并强制可观察到的元素在没有任何订阅服务器的情况下生成事件。
在没有任何订阅者的情况下强制订阅 – connect() 在 subscribe() 之前
为了证明这一点，让我们更正我们的例子：
public static void connectBeforeSubscribe() throws InterruptedException {
ConnectableObservable obs = getObservable()
.doOnNext(x -> LOGGER.info(“saving ” + x)).publish();
LOGGER.info(“connecting:”);
Subscription s = obs.connect();
Thread.sleep(1000);
LOGGER.info(“subscribing #1”);
obs.subscribe((i) -> LOGGER.info(“subscriber#1 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“subscribing #2”);
obs.subscribe((i) -> LOGGER.info(“subscriber#2 is printing x-coordinate ” + i));
Thread.sleep(1000);
s.unsubscribe();
}
步骤相对简单：
首先, 我们连接
然后我们等待一秒钟并订阅第一个订阅者
最后，我们等待另一秒钟并订阅第二个订阅者
请注意，我们添加了 doOnNext() 运算符。这里我们可以在数据库中存储元素，例如在我们的代码中，我们只打印“save…”。
如果我们启动代码并开始点击，我们将看到在 connect() 调用之后立即发出和处理元素：
connecting:
saving 306
saving 248
subscribing #1
saving 377
subscriber#1 is printing x-coordinate 377
saving 295
subscriber#1 is printing x-coordinate 295
saving 206
subscriber#1 is printing x-coordinate 206
subscribing #2
saving 347
subscriber#1 is printing x-coordinate 347
subscriber#2 is printing x-coordinate 347
clearing resources
如果没有订阅者，则仍会处理这些元素。
因此，不管是否有人订阅，connect() 方法都会开始发出和处理元素，就好像有一个使用了元素的空操作的人工订阅器一样。
如果有一些真正的订阅者订阅，这个人工中介只向他们传播元素。
若要取消订阅，我们会执行以下步骤:
s.unsubscribe();
然后：
Subscription s = obs.connect();
autoConnect()
此方法意味着在订阅之前或之后不会调用 connect()，而是在第一个订阅者订阅时自动调用。
使用此方法，我们不能自己调用 connect(), 因为返回的对象是通常的 Observable，它没有此方法但使用底层的 ConnectableObservable：
public static void autoConnectAndSubscribe() throws InterruptedException {
Observable obs = getObservable()
.doOnNext(x -> LOGGER.info(“saving ” + x)).publish().autoConnect();
LOGGER.info(“autoconnect()”);
Thread.sleep(1000);
LOGGER.info(“subscribing #1”);
Subscription s1 = obs.subscribe((i) ->
LOGGER.info(“subscriber#1 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“subscribing #2”);
Subscription s2 = obs.subscribe((i) ->
LOGGER.info(“subscriber#2 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“unsubscribe 1”);
s1.unsubscribe();
Thread.sleep(1000);
LOGGER.info(“unsubscribe 2”);
s2.unsubscribe();
}
请注意，我们也不能取消订阅人工订阅者。我们可以取消订阅所有真正的订阅者，但人工订阅者仍将处理事件。
为了理解这一点，让我们看一下最后一个订阅者取消订阅后最后发生的事情：
subscribing #1
saving 296
subscriber#1 is printing x-coordinate 296
saving 329
subscriber#1 is printing x-coordinate 329
subscribing #2
saving 226
subscriber#1 is printing x-coordinate 226
subscriber#2 is printing x-coordinate 226
unsubscribe 1
saving 268
subscriber#2 is printing x-coordinate 268
saving 234
subscriber#2 is printing x-coordinate 234
unsubscribe 2
saving 278
saving 268
正如我们所看到的，在第二次取消订阅后，不会出现清除资源的情况，并继续使用 doOnNext() 保存元素。这意味着人工订阅服务器不会取消订阅，而是继续使用元素。
refCount()
refCount() 类似于 autoConnect()，因为只要第一个订阅者订阅，连接也会自动发生。
与 autoconnect() 不同，当最后一个订阅者取消订阅时，也会自动断开连接：
public static void refCountAndSubscribe() throws InterruptedException {
Observable obs = getObservable()
.doOnNext(x -> LOGGER.info(“saving ” + x)).publish().refCount();
LOGGER.info(“refcount()”);
Thread.sleep(1000);
LOGGER.info(“subscribing #1”);
Subscription subscription1 = obs.subscribe(
i -> LOGGER.info(“subscriber#1 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“subscribing #2”);
Subscription subscription2 = obs.subscribe(
i -> LOGGER.info(“subscriber#2 is printing x-coordinate ” + i));
Thread.sleep(1000);
LOGGER.info(“unsubscribe#1”);
subscription1.unsubscribe();
Thread.sleep(1000);
LOGGER.info(“unsubscribe#2”);
subscription2.unsubscribe();
}
refcount()
subscribing #1
saving 265
subscriber#1 is printing x-coordinate 265
saving 338
subscriber#1 is printing x-coordinate 338
subscribing #2
saving 203
subscriber#1 is printing x-coordinate 203
subscriber#2 is printing x-coordinate 203
unsubscribe#1
saving 294
subscriber#2 is printing x-coordinate 294
unsubscribe#2
clearing resources
结论
ConnectableObservable 类可以轻松地处理多个订阅者。
它的方法看起来很相似，但由于实现上的细微差别 (甚至方法的顺序也很重要)，用户的行为发生了很大的变化。