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())，否則你會在崩潰告終。