自定义筛选器运算符RxJava

Question

我试图拥抱RxJava的荣耀并将其整合到我的应用程序中。我已经编写了下面的代码来添加漫画，其累积成本不超过定义的预算。为了实现这一点，我写了2个实现。

1. 用途Observable.create()这与订阅和背压
2. 用途已有运营商在RxAndroid LIB气馁，因为复杂的居多。

我很想得到的反馈上的实现是在性能，内存消耗和简单性方面一个更好的，如果把订阅和背压Observable.create()处理静置片刻？

第一个实现：

Observable<Integer> filterObservable = Observable.create(new ObservableOnSubscribe<Integer>() { 
    @Override 
    public void subscribe(ObservableEmitter<Integer> e) throws Exception { 
     Timber.d("filterComicsAccordingToBudget():subscribe"); 
     int pageCountOfComicsWithInBudget = 0; 
     double totalCost = 0.0; 
     for(MarvelComic comic : getMarvelComicsList()) { 
      totalCost += Double.valueOf(comic.getPrice()); 
      Timber.d("totalCost: %s budget: %s priceOfComic: %s", totalCost, budget, comic.getPrice()); 
      if(totalCost > budget) { 
       break; 
      } 
      pageCountOfComicsWithInBudget += Integer.valueOf(comic.getPageCount()); 
      Timber.d("pageCount: %s price: %s comicName: %s totalPages: %s", comic.getPageCount(), comic.getPrice(), comic.getTitle(), pageCountOfComicsWithInBudget); 
      e.onNext(pageCountOfComicsWithInBudget); 
     } 
     e.onComplete(); 
    } 
}); 

filterObservable.subscribeOn(Schedulers.computation()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Observer<Integer>() { 
     int comicCount = 0; 
     int pageCountOfComicsWithInBudget = 0; 

     @Override 
     public void onSubscribe(Disposable d) { 
      Timber.d("filterComicsAccordingToBudget():onSubscribe"); 
     } 

     @Override 
     public void onNext(Integer pageCountOfComicsWithInBudget) { 
      Timber.d("filterComicsAccordingToBudget():onNext"); 
      comicCount++; 
     } 

     @Override 
     public void onError(Throwable e) { 
      Timber.e("onFilterComicsForBudget:onError() %s", e); 
     } 

     @Override 
     public void onComplete() { 
      Timber.d("filterComicsAccordingToBudget():onComplete"); 
     } 
    } 
}); 

第二个执行：

Observable.fromIterable(getMarvelComicsList()) 
    .map(new Function<MarvelComic, HashMap<String, Double>>() { 
     HashMap<String, Double> myMap = new HashMap<String, Double>(); 
     double count = 0; 

     @Override 
     public HashMap<String, Double> apply(@NonNull MarvelComic marvelComic) throws Exception { 
      myMap.put("price", Double.valueOf(marvelComic.getPrice())); 
      myMap.put("pageCount", Double.valueOf(marvelComic.getPageCount())); 
      myMap.put("comicsCount", count++); 
      return myMap; 
     } 
    }) 
    .takeWhile(new Predicate<HashMap<String, Double>>() { 

     double sum; 

     @Override 
     public boolean test(@NonNull HashMap<String, Double> map) throws Exception { 
      Timber.e("sum is: %s", sum); 
      return (sum += map.get("price")) < 5.00; 
     } 
    }) 
    .subscribe(new Observer<HashMap<String, Double>>() { 

     @Override 
     public void onSubscribe(Disposable d) { 
     } 

     @Override 
     public void onNext(HashMap<String, Double> map) { 
      Timber.e("value in onNext is: %s %s %s", map.get("pageCount"), map.get("price"), map.get("comicsCount")); 
     } 

     @Override 
     public void onError(Throwable e) { 
      Timber.e("onError()!!! %s",e); 
     } 

     @Override 
     public void onComplete() { 
      Timber.e("onComplete()!!!"); 
     } 
    }); 

我还挺花哨的第一个实现，因为它更迫切，我已经习惯了，似乎不太笨重给我但考虑到我在RxJava中的有限知识，我可能完全错误。

Answer 1

我会避免为这类操作创建自定义Observable。您可以使用普通的RxJava操作员完成所有需要的操作。

在飞，我会做这样的事情：

private Observable<Double> getLimitObservable(final double budget) { 
    return Observable.fromIterable(getMarvelComicsList()) 
      .scan(0D, (aDouble, marvelComic) -> aDouble + marvelComic.getPrice()) 
      .takeWhile(aDouble -> aDouble < budget) 
      .skip(1); 
} 

上面的代码使用scan（也叫蓄电池）运营商保持漫画的价格总额的轨道。更多详情here。所以，现在从新的Observable返回一个double（代表总数）。在此之后，我们有takeWhile停止排放的项目，直到条件保持真实。最终我跳过第一个项目，因为上面提到的Observable将会发出至少一个项目（在条件可以验证之前）。

Observable.zip(getLimitObservable(500d), Observable.fromIterable(getMarvelComicsList()), (aDouble, marvelComic) -> marvelComic) 
       .subscribe(marvelComic -> Log.d("test", "comic: " + marvelComic.getName())); 

现在我结合以前观察到的一个新（使用zip运营商），将产生对每对夫妇的项目（一个来自第一观察到，和一个从第二）一个新的项目，在这方式你会得到一些项目等于从两个observables发出的最小数量的项目。更多详细信息here

这将打印列表中的第一部漫画，直到达到预算限制。

我敢打赌，有更好的解决方案，但这只是一个例子。