RxJava，一个可以观察到的多个订阅者：publish（）.autoconnect（）

Question

Observable<Integer> dataStream = Observable.just(1, 2, 3).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());

Log.d(TAG, "subscribing sub1...");
dataStream.subscribe(v -> System.out.println("Subscriber #1: "+ integer));

Log.d(TAG, "subscribing sub2...");
dataStream.subscribe(v -> System.out.println("Subscriber #2: "+ integer));

D/RxJava: subscribing sub1...
D/RxJava: subscribing sub2...
D/RxJava: Subscriber #1: 1
D/RxJava: Subscriber #1: 2
D/RxJava: Subscriber #1: 3
D/RxJava: Subscriber #2: 1
D/RxJava: Subscriber #2: 2
D/RxJava: Subscriber #2: 3

我是这样理解的，从可观察的角度来看：

>

有人订阅了我，我应该开始发送项目
[订阅者:1][要发送的项目：1,2,3]

向订阅服务器发送项“1”
[订阅服务器：1][要发送的项：2,3]

...

但它不是这样运作的。就像它们是两个独立的可观测物在一个。这让我很困惑，为什么他们不把项目给所有的订户？

奖金：

谢了！

Answer 1

这是因为事实上这是两个独立的可观察物。它们是在调用subscribe()时“生成”的。因此，你所提供的步骤在意义上是不正确的，即步骤3和4只是1和2，但在一个不同的可观察的。

但是，由于日志记录发生在线程上，您将它们视为1 1 1 2 2 2。如果您要删除observeon()部分，那么您将看到以交织的方式发送。要查看下面的运行代码：

@Test
public void test() throws InterruptedException {
    final Scheduler single = Schedulers.single();
    final long l = System.nanoTime();
    Observable<Long> dataStream =
            Observable.just(1, 2, 3)
                    .map(i -> System.nanoTime())
                    .subscribeOn(Schedulers.computation());
                    //.observeOn(single);

    dataStream.subscribe(i -> System.out.println("1  " + Thread.currentThread().getName() + " " + (i - l)));
    dataStream.subscribe(i -> System.out.println("2  " + Thread.currentThread().getName() + " " + (i - l)));

    Thread.sleep(1000);
}

输出，至少在我的运行中是（请注意线程名）：

1  RxComputationThreadPool-1 135376988
2  RxComputationThreadPool-2 135376988
1  RxComputationThreadPool-1 135486815
2  RxComputationThreadPool-2 135537383
1  RxComputationThreadPool-1 135560691
2  RxComputationThreadPool-2 135617580

1  RxSingleScheduler-1 186656395
1  RxSingleScheduler-1 187919407
1  RxSingleScheduler-1 187923753
2  RxSingleScheduler-1 186656790
2  RxSingleScheduler-1 187860148
2  RxSingleScheduler-1 187864889

@Test
public void test() throws InterruptedException {
    final Scheduler single = Schedulers.single();
    final long l = System.nanoTime();
    ConnectableObservable<Long> dataStream =
            Observable.just(1, 2, 3)
                    .map(i -> System.nanoTime())
                    .subscribeOn(Schedulers.computation())
                    .observeOn(single)
                    .publish();

    dataStream.subscribe(i -> System.out.println("1  " + (i - l)));
    dataStream.subscribe(i -> System.out.println("2  " + (i - l)));

    Thread.sleep(1000);
    dataStream.connect();
    Thread.sleep(1000);

}

您将注意到，在第一秒钟（第一次thread.sleep()调用）没有发生任何事情，而在datastream.connect()被调用之后，就发生了发射。

refcount()接受ConnectableObservable，并通过计算当前订阅了多少订阅服务器来向订阅服务器隐藏调用Connect()的需要。它所做的是在第一次订阅时调用connect()，在最后一次取消订阅后取消对原始可观察到的订阅。

至于相互取消publish().autoconnect()，之后您确实得到了一个可观察的，但它有一个特殊的属性，假设最初的可观察的通过Internet进行API调用（持续10秒），当您在不使用share()的情况下使用它时，对服务器的并行查询将与这10秒内的订阅数量一样多。另一方面，对于share()，只有一次调用。

如果共享的可观察到的工作完成得非常快（就像(1,2,3)),您将看不到它的任何好处。

autoconnect()/refcount()为您提供了一个您订阅的中间可观测项，而不是原始可观测项。

如果您感兴趣，请阅读这本书：用RxJava进行反应性编程