RXJavaのバックグラウンドスレッドで長い計算を実行する方法

Question

アンドロイドでRXJavaを使用して、バックグラウンドスレッドで長い計算を実行したいとします。計算後、私は結果をRecylerviewに提示しようとしています。 300 items.IがRXJavaに新しいです、その後のArrayList eventFeedItemsサイズは詳細であると私はUI Hindringに直面しています上記のコード切れ端で

Observable.just("true") 
 
       .subscribeOn(Schedulers.io()) 
 
       .map(new Func1<String, String>() { 
 
        @Override 
 
        public String call(String s) { 
 
         feedlist.clear(); 
 
         if (eventFeedItems != null && !eventFeedItems.isEmpty()) { 
 
          for (int i = 0; i < eventFeedItems.size(); i++) { 
 
           if (eventFeedItems != null && eventFeedItems.get(i) != null 
 
             && ((eventFeedItems.get(i).getType() != null && eventFeedItems.get(i).getType().equalsIgnoreCase("EVENT")) 
 
             || (eventFeedItems.get(i).getActivityRequestType() != null && eventFeedItems.get(i).getActivityRequestType().equalsIgnoreCase(EventConstants.TRENDING_ACTIVITY)))) { 
 
            if (eventFeedItems.get(i).getActivityRequestType() != null && !eventFeedItems.get(i).getActivityRequestType().equalsIgnoreCase("")) { 
 
             feedlist.add(new FeedsListModel(eventFeedItems.get(i), eventFeedItems.get(i).getActivityRequestType(), null)); 
 
            } else if (eventFeedItems.get(i).getRequestType() != null && !eventFeedItems.get(i).getRequestType().equalsIgnoreCase("")) { 
 
             feedlist.add(new FeedsListModel(eventFeedItems.get(i), eventFeedItems.get(i).getRequestType(), null)); 
 
            } else 
 
             feedlist.add(new FeedsListModel(eventFeedItems.get(i), EventConstants.ATTENDEE_POST, null)); 
 
           } 
 
          } 
 
         } 
 
         Log.d("calculations","Completed"); 
 
         return ""; 
 
        } 
 
       }) 
 
       .observeOn(AndroidSchedulers.mainThread()) 
 
       .subscribe(new Action1<String>() { 
 
        @Override 
 
        public void call(String s) { 
 
//      feed_list.setLayoutManager(mLayoutManager); 
 
//      feedListAdapter.notifyDataSetChanged(); 
 
         Log.d("Adapter", "Set"); 
 
        } 
 
       }, new Action1<Throwable>() { 
 
        @Override 
 
        public void call(Throwable throwable) { 
 
         Log.d("Exception", "oh! fish..."); 
 
         throwable.printStackTrace(); 
 
        } 
 
       });

：私は、次のコードを使用しています。私を助けてください。

Answer 1

map-Operatorを使用して並行性を実現することはできません。

最初のsubscribeOnは、すべてのエミッションをIOスケジューラに移動します。並行処理はここでは起こりません。

.subscribeOn(Schedulers.io()) 

map-operatorは、前のスレッドから同期して呼び出されます。あなたの場合は、IOスレッドプールからいくつかのスレッドになります。

.map(new Func1<String, String>() { 

マップ-オペレータが実行された後の値から変換された後、あなたは

.observeOn(AndroidSchedulers.mainThread()) 

でのAndroid-UI-イベントループにIO-スレッドから値を移動しますUIスレッドへのIOスレッドは、最初の観測値からの次の値が処理されます。

Observable.just("true") 

この例では、1つの値しか生成されないため、これ以上の値はありません。

同時性を実現するには、mapの代わりにflatMapを使用する必要があります。また、フラットマップでsubscribeOn（）を使用して、別のスレッドで各ストリームを作成します。

どのように並行処理が行われているかを確認するには、この例を検討してください。観察可能なものはすべて同時に購読されるので、テゼットの時間は約5秒です。同時実行が発生すると、1 + 2 + 3 + 4 + 5秒に実行時間を加えた時間がかかります。

@Test 
public void name1() throws Exception { 

     Observable<Integer> value = Observable.just(1_000, 2_000, 3_000, 4_000, 5_000) 
       .flatMap(i -> Observable.fromCallable(() -> doWork(i)).subscribeOn(Schedulers.io()) 
       ).doOnNext(integer -> System.out.println("value")); 

     value.test().awaitTerminalEvent(); 
} 

private int doWork(int sleepMilli) { 
     try { 
      Thread.sleep(sleepMilli); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 

     return -1; 
} 

あなたは同時実行がflatMapで起こっているかについての詳細をお知りになりたい場合は、私はお勧めしますあなたのコードに関してはhttp://tomstechnicalblog.blogspot.de/2015/11/rxjava-achieving-parallelization.html

を読んでみてください。

• 移動インタフェースのanonymouse実装へのプライベート内部クラスの実装とそのインスタンスを使用します。あなたはより読みやすい観察を得るでしょう。

• オペレータのグローバル変数に副作用を使わないでください。同時実行が関係する場合、 は競合状態になります。

  List<FeedsListModel> eventFeedItems = Arrays.asList(new FeedsListModel(), new FeedsListModel()); 
  
  Observable<FeedsListModel> feedsListModelObservable = Observable.fromIterable(eventFeedItems) 
          .flatMap(feedsListModel -> Observable.fromCallable(() -> calculation(feedsListModel)) 
            .subscribeOn(Schedulers.computation()) 
  ); 
  
  feedsListModelObservable 
       .toList() 
       .observeOn(Schedulers.io()) 
       .subscribe(feedsListModels -> { 
        // do UI stuff 
  }); 
  

支援法：

private FeedsListModel calculation(FeedsListModel model) { 
    // do calculation here 

    return new FeedsListModel(); 
}