如何在RXJava中对后台线程执行长时间计算Android

Question

我想在Android中使用RXJava在后台线程上执行长时间计算。经过计算，我试图在Recylerview中呈现结果。我使用下面的代码段：

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

利用上述片的我面对UI Hindring码作为该ArrayList eventFeedItems尺寸在大约多于300 items.I相信新的RXJava 。请帮助我。

Answer 1

你不会使用map-Operator实现并发。

第一个subscribeOn会将所有想法移动到IO调度程序。没有并发发生在这里。

.subscribeOn(Schedulers.io()) 

map-operator将从前一个线程同步调用。在你的情况下，它将来自IO-threadpool的一些线程。

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

地图运营商已执行完毕后，你就会从IO线程到Android的UI事件循环的价值与

.observeOn(AndroidSchedulers.mainThread()) 

值已经从翻译后移动IO线程到UI线程将处理来自初始可观察值的下一个值。

Observable.just("true") 

在你的例子中，将不会有更多的值，因为你只产生一个值。

为了实现并发性，您应该使用flatMap而不是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(); 
}