rxjava zip在指定的调度程序上不工作？

Question

我有一个实现，其中我有两个单线程，它们是对数据库的并行调用，以获取对象。然后我使用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函数仍然在默认的主线程中运行，而我希望它在计算线程中。不知道为什么会这样，有什么想法吗？

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

Answer 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();
    };
}