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是当前活动正在实现的接口）。

我在使用map（I）的线路上遇到错误-

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

内置。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（）），否则会导致崩溃。