rxJava 如何让 flatMap 在多线程上运行

Question

我希望从flatMap发出的每个项目都在自己的线程上运行

这是一个简化的实际用法示例，其中每个项目都将是一个url请求.
Add subscribeOn(Schedulers.io()) 在每个单线程上仍然在单个线程上运行
这里的规则是什么？

Integer[] array= new Integer[100];
for (int i = 0; i < 100; i++){
array[i] = i+1;
}
Observable.fromArray(array)
.flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
@Override
public SingleSource<Integer> apply(Integer i) throws Throwable {
Log.i(TAG, "apply " +  i + " " + Thread.currentThread().getName());
return Single.just(i).subscribeOn(Schedulers.io());
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Integer i) {
// Log.i(TAG, "onNext " + Thread.currentThread().getName() + i);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
}
});

结果：

2020-12-16 22:54:47.010 10649-10700/com.example.rxjava I/MYTAG: apply 1 RxCachedThreadScheduler-1
2020-12-16 22:54:47.037 10649-10700/com.example.rxjava I/MYTAG: apply 2 RxCachedThreadScheduler-1
2020-12-16 22:54:47.038 10649-10700/com.example.rxjava I/MYTAG: apply 3 RxCachedThreadScheduler-1
2020-12-16 22:54:47.039 10649-10700/com.example.rxjava I/MYTAG: apply 4 RxCachedThreadScheduler-1
2020-12-16 22:54:47.040 10649-10700/com.example.rxjava I/MYTAG: apply 5 RxCachedThreadScheduler-1
2020-12-16 22:54:47.043 10649-10700/com.example.rxjava I/MYTAG: apply 6 RxCachedThreadScheduler-1
2020-12-16 22:54:47.051 10649-10700/com.example.rxjava I/MYTAG: apply 7 RxCachedThreadScheduler-1
2020-12-16 22:54:47.051 10649-10700/com.example.rxjava I/MYTAG: apply 8 RxCachedThreadScheduler-1

Answer 1

除了使用just之外，您走在正确的轨道上，它采用现有对象，因此无论之前创建和计算该对象发生什么。在这种情况下，它是从同一线程调用的flatMapSingle的 lambda。

您必须使计算本身成为流的一部分，以便通过fromCallable并行运行，例如：

Observable.fromArray(array)
.flatMapSingle(i -> {
return Single.fromCallable(() -> {
Log.i(TAG, "apply " +  i + " " + Thread.currentThread().getName());
return i + 1000;
})
.subscribeOn(Schedulers.io());
})
.observeOn(AndroidSchedulers.mainThread())
// ...
;