rxjava zip 不适用于指定的调度程序？答案

【问题标题】：rxjava zip not working on specified schedulers?rxjava zip 不适用于指定的调度程序？
【发布时间】：2021-05-06 19:05:15
【问题描述】：

我有这个实现，我有 2 个单曲，它们是对数据库的并行调用以获取对象。然后我使用 rxjava zip 运算符来聚合结果。如下：

...
(other code)

Single<A> a = Single.fromCallable(() -> {
        System.out.println("Inside single a, running in thread " + Thread.currentThread().getName());
}).subscribeOn(Schedulers.io);

Single<B> b = Single.fromCallable(() -> {
        System.out.println("Inside single b, running in thread " + Thread.currentThread().getName());
}).subscribeOn(Schedulers.io);

Single<C> result = Single.zip(a, b, aggregationFunc2())
    .subscribeOn(Schedulers.computation())
    .subscribe(response::resume, response::resume);

private Func2<a, b, c> aggregationFunc2() {
        System.out.println("Inside aggregationFunc2, running in thread " + Thread.currentThread().getName());
        return (a, b) -> {
            CBuilder cBuilder = new Builder();
            if (a != null) {
                cBuilder.setA(a);
            }
            if (b != null) {
                cBuilder.setB(b);
            }           
            return cBuilder.build();
        };
    }

但是，我得到的日志是 a 和 b 都在具有不同 io 线程的 IO 线程池中运行，例如 io-2 和 io-3。这是意料之中的，因为我指定每个单曲都在 IO 调度程序中。但是 zip 函数仍在默认主线程中运行，而我希望它在计算线程中。不知道为什么会这样，有什么想法吗？

另外，我尝试了 zip 的 observeOn()，它仍然在默认的主线程中运行。

【问题讨论】：

标签： java observable jax-rs rx-java reactive-programming

【解决方案1】：

因为你在主线程上调用了aggregationFunc2。

稍微重写你的代码应该会发现误解：

Func2<A, B, C> f = aggregationFunc2();
Single<C> result = Single.zip(a, b, f);

在创建和返回函数本身之前打印。您必须在创建的函数中移动 print 语句：

private Func2<a, b, c> aggregationFunc2() {
    return (a, b) -> {
        System.out.println("Inside aggregationFunc2, running in thread " 
             + Thread.currentThread().getName());
        CBuilder cBuilder = new Builder();
        if (a != null) {
            cBuilder.setA(a);
        }
        if (b != null) {
            cBuilder.setB(b);
        }           
        return cBuilder.build();
    };
}

【讨论】：