java - 如何在 rxJava 中订阅不同的线程池

Question

我现在正在我的程序中运行这个位

Observable.from(constituents.entrySet()).subscribeOn(Schedulers.newThread())
        .flatMap(Async.toAsync((Map.Entry<String, ConstituentInfo> entry) -> {
            logger.info(Thread.currentThread().getName());
            ConstituentInfo constituent = entry.getValue();

            String securityBySymbol = Redux.getSecurityBySymbol(entry.getKey());

            String company = UtilityMethods.getNestedJsonObject(securityBySymbol, "company");
            Integer compId = UtilityMethods.getIntegerFromJsonObject(company, "id");
            String companyName = UtilityMethods.getStringFromJsonObject(company, "name");
            String tier = UtilityMethods.getNestedJsonObject(securityBySymbol, "tier");
            String tierId = UtilityMethods.getStringFromJsonObject(tier, "id");
            String marketPlace = UtilityMethods.getStringFromJsonObject(tier, "name");
            String countryName = getCountryName(compId);

            constituent.setCompanyName(StringUtils.isBlank(companyName) ? NA : companyName);
            constituent.setMarketPlace(StringUtils.isBlank(marketPlace) ? NA : marketPlace);
            constituent.setCountryName(StringUtils.isBlank(countryName) ? NA : countryName);
            constituent.setTierId(StringUtils.isBlank(tierId) ? NA : tierId);

            return constituent;
        })).subscribeOn(Schedulers.newThread())
        .toList()
        .timeout(30, TimeUnit.MINUTES)
        .toBlocking()
        .single();

并且它是并发运行的，但是它运行在RxComputationThreadPool上。我想知道如何让它在 Schedulers.newThread() 上运行，以及它是否会提高性能。

或者，如果它不能提高性能，有没有办法让下面的代码运行得更快？

Answer 1

toAsync 存在重载，它需要 Scheduler，而您不需要 subscribeOn。 computation() 调度程序是所有调度程序中延迟最低的。 io() 很可能，而 newThread() 肯定会启动一个新线程，因此可能需要几百微秒来执行第一个任务，但它们非常适合阻塞 I/O或网络调用，其中延迟并不重要。