与 RxJava 并行处理 IO 密集型任务列表

Question

我需要并行获取一些文件。 get 操作本身是 IO 密集型的，可以从并行执行中受益匪浅。

使用 RxJava，我可以通过使用 Async.toAsync 包装我的函数来实现这一点。

我想知道使用subscribeOn() 或observeOn() 是否有更简洁的方法？我无法弄清楚。尝试了不同的方式，但任何方式都只能使用一个线程，并且处理是按顺序进行的。

import rx.Observable;
import rx.Scheduler;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import rx.util.async.Async;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class ParallelMultiGet {

    public List<String> readContents(List<String> paths) {

        Func1<String, String> getFunction = new Func1<String, String>() {
            @Override
            public String call(String path) {
                return get(path);
            }
        };

        Scheduler scheduler = Schedulers.from(
                Executors.newFixedThreadPool(
                        Math.min(paths.size(), 50)));

        Future<List<String>> result = Observable
                .from(paths)
                .flatMap(
                        Async.toAsync(getFunction, scheduler))
                .toList()
                .toBlocking()
                .toFuture();
        try {
            return result.get(30, TimeUnit.MINUTES);
        } catch (Exception e) {
            // For example if Func1.call above throws an exception it ends up in here
            throw new IllegalStateException("Couldn't read paths", e);
        }
    }

    private String get(String path) {
        // this would be the slow operation, waiting for IO
        return "content";
    }

}

即使这已经很不错了，因为我不需要构建自己的循环来提交期货并将它们中的值组合到结果列表中。但也许不必这么冗长？

Answer 1

为了不那么冗长：

• 你可以使用 RxJava 中的超时操作符
• 您可以将 java8 与 lambda 一起使用，而不是 Func1（但对您的情况没有多大帮助）
• 您可以避免创建自己的执行程序

我得到了应该和你做同样事情的代码：

  public List<String> readContents(List<String> paths) {

      try {

        return Observable
                .from(paths)
                .flatMap(Async.toAsync((Func1<String, String>) this::get, Schedulers.io()))
                .toList()
                .timeout(30, TimeUnit.MINUTES)
                .toBlocking().single();
      } catch (RuntimeException ex) { // the cause will be a timeoutException
        // For example if Func1.call above throws an exception it ends up in here
        throw new IllegalStateException("Couldn't read paths", ex);
      }
    }

    private String get(String path) {
        // this would be the slow operation, waiting for IO
        return "content";
   }