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); 

里面的方法完成（）我正在更新GUI（finishedListener的是，目前活动正在实施一个接口）。

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

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

compile 'io.reactivex:rxjava:1.2.9' 

- （> intenseCalculation（I）I）

我正在与地图的线得到一个错误

我该如何解决这个问题？

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())，否则你会在崩溃告终。