如何用RxJava并行转换Single<Set<String>>为Single<Set<Integer>>？

嗨，这个问题我熟！要实现并行转换每个String到Integer，最后合并成Set并包装回Single，其实用RxJava的几个操作符就能搞定，我给你两种靠谱的方案：

方案一：用flatMap+并行调度实现 #

这种方式适合需要精细控制每个元素线程的场景，代码如下：

import io.reactivex.Single;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import java.util.HashSet;
import java.util.Set;

// 你的原始Single<Set<String>>
Single<Set<String>> sss = Single.just(Sets.newHashSet("1", "2", "3"));

// 转换为Single<Set<Integer>>
Single<Set<Integer>> ssi = sss
    // 把Single里的Set拆成逐个发射String的Observable
    .flatMapObservable(set -> Observable.fromIterable(set))
    // 每个String单独开线程并行转换
    .flatMap(str -> Observable.just(str)
        .subscribeOn(Schedulers.io()) // 用IO线程池处理高开销转换
        .map(Integer::parseInt))
    // 把所有转换后的Integer收集到HashSet里
    .collect(HashSet::new, Set::add)
    // 转成Single类型
    .toSingle();

方案二：用ParallelFlowable简化并行逻辑（RxJava 2+支持） #

如果你的RxJava版本是2及以上，用ParallelFlowable会更直观，它专门用来处理并行任务：

import io.reactivex.Single;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import java.util.HashSet;
import java.util.Set;

Single<Set<String>> sss = Single.just(Sets.newHashSet("1", "2", "3"));

Single<Set<Integer>> ssi = sss
    .flatMapObservable(Observable::fromIterable)
    .parallel() // 转为并行流
    .runOn(Schedulers.io()) // 指定并行执行的线程池
    .map(Integer::parseInt) // 并行转换每个元素
    .sequential() // 合并回普通Observable
    .collect(HashSet::new, Set::add)
    .toSingle();

关键说明 #

• 为什么不用你之前的flatMapIterable+map？因为那样是串行执行转换的，没法利用多线程并行。我们通过flatMap（或parallel）让每个转换任务跑在独立线程上，才符合你“并行转换”的要求。
• 关于线程安全：collect操作是在下游单线程执行收集的，所以普通的HashSet就够用，不需要线程安全容器。
• 如果担心转换失败（比如String不是合法数字），可以在map后面加onErrorReturn或onErrorResumeNext来处理异常情况。

内容的提问来源于stack exchange，提问作者nan