我们可以在调度程序上运行的可观测数据的数量有限制吗？

Question

我正在尝试查看是否可以在IO()和Computation()调度程序上生成100万个可观察的。

 public static void observableLimit() {
        sum = 0;
        long lowerBound = 0;
        long higherBound = 1000;
        Flowable.fromCallable(() -> {
            Flowable.rangeLong(lowerBound, higherBound + 1)
                    .subscribe(integer -> Observable.just(integer)
                            .subscribeOn(Schedulers.io())
                            .subscribe(j -> {
                                printNum(j);
                                sum = sum + j;
                            }));
            return true;
        }).blockingSubscribe(aBoolean -> {
            long actualSum = (higherBound * (higherBound + 1)) / 2;
            System.out.println("");
            System.out.println("SUM: " + sum);
            Assert.assertEquals(actualSum, sum);
        });
    }

对于higherbound=100来说，它大部分时间都可以工作，对于1000来说，它有时可以工作，但大部分时间都失败，对于10000来说，它几乎每次都失败，如果我告诉它在newthread()上运行它，并且根本不使用subscribeon()的话，它就可以工作。

我怎样才能纠正这种行为呢？

Answer 1

您面临的问题不是关于可观察性的某些限制，而是您的代码的问题。您的blockingsubscribe是对一个与跨所有其他线程的可流无关的可流的blockingsubscribe。对于HigherBound的小值，您将看到代码可以工作，而对于大值则不行，这是因为对于小的HigherBound来说，外部可流动可能与内部可流动一样快，但是对于HigherBound的高值来说，折叠得更快。

我想说的是，为了看到正确的结果，您需要与跨所有其他线程而不是外部线程的可流同步。我还将long sum替换为线程安全实现longadder sum，您可以使用flatmap运算符来实现。

Flowable.rangeLong(lowerBound, higherBound + 1)
         .flatMap(t -> Flowable.just(t)
                 .subscribeOn(Schedulers.io())
         )
         .doOnNext(sum::add)
         .doOnComplete(() -> {
             long actualSum = (higherBound * (higherBound + 1)) / 2;
             log("SUM: " + sum.longValue() + ", ACTUAL: " + actualSum);
             log("Equals: " + (actualSum == sum.longValue()));
         })
         .blockingSubscribe();