Android RxJava 并行化 - 使用 ExecutorService 进行计算

Question

我已经用 RxJava 成功地完成了一个小型 Java 程序。代码是：

    public static void main( String[] args ) {
        int threadCt = Runtime.getRuntime().availableProcessors() + 1;
        //multi-threading
        ExecutorService executor = Executors.newFixedThreadPool(threadCt);
        Scheduler scheduler = Schedulers.from(executor);

        final AtomicInteger batch = new AtomicInteger(0);

        Observable.range(1,80)
            .groupBy(i -> batch.getAndIncrement() % threadCt )
            .flatMap(g -> g.observeOn(scheduler)
                    .map(i -> intenseCalculation(i))
            ).subscribe(System.out::println);
    }



    public static int intenseCalculation(int i) {
       try {
          System.out.println("Calculating " + i +
                " on " + Thread.currentThread().getName());
          Thread.sleep(500);
          return i;
       } catch (InterruptedException e) {
        throw new RuntimeException(e);
       }
    }

使用此代码一切正常。现在我正在尝试将此代码传递给 Android：

        Scheduler scheduler = Schedulers.from(executor);

        final AtomicInteger batch = new AtomicInteger(0);
        Observable.range(0, copiedCategories.size() - 1)
                .groupBy(i -> batch.getAndIncrement() % threadCt)
                .flatMap(g -> g.observeOn(scheduler))
                .map(i -> intenseCalculation(i))
                .subscribe(finishedListener::finished);

在finished() 方法中，我正在更新GUI（finishedListener 是当前Activity 正在实现的接口）。

我在使用 map(i -> strongCalculation(i)) 时遇到错误：

no instance(s) of type variable(s) exist so that void conforms to R

在我正在使用的 build.gradle（用于应用程序）中：

compile 'io.reactivex:rxjava:1.2.9'

我该如何解决这个问题？

Answer 1

在java代码中你有：

.flatMap(g -> g.observeOn(scheduler)
               .map(i -> intenseCalculation(i))
        )

但在 Android 代码中，您在主流上执行 map()：

.flatMap(g -> g.observeOn(scheduler))
.map(i -> intenseCalculation(i))

所以，这些不是一回事，Android 代码中有额外的括号，这完全改变了流逻辑。

注意，您应该在UI线程上执行UI相关的操作，所以您必须在订阅流之前执行observeOn(AndroidSchedulers.mainThread())，否则您最终会崩溃。