0

我使用sqlbrite来监听表a和b的变化。并使用 combineLatest 运算符来组合 sqlbrite 生成的 observables。在 BiFunction 过程中发出 observableA 和 observableB 的项。

private CompositeDisposable mSubscriptions = new CompositeDisposable();
private void initialize(){
    QueryObservable observableA = mDb.createQuery("table_a", "select * from table_a", null);
    QueryObservable observableB = mDb.createQuery("table_b", "select * from table_b", null);
    ResourceSubscriber subscriber = Flowable.combineLatest(
            RxJavaInterop.toV2Observable(observableA
                    .mapToList(mTableAMapperFunction)).toFlowable(BackpressureStrategy.LATEST)
            , 
            RxJavaInterop.toV2Observable(observableB
                    .mapToList(mTableBMapperFunction)).toFlowable(BackpressureStrategy.LATEST)
            , new BiFunction<List<ItemA>, List<ItemB>, List<ResultItem>>() {
                @Override
                public List<ResultItem> apply(@io.reactivex.annotations.NonNull List<ItemA> aItems, @io.reactivex.annotations.NonNull List<ItemB> bItems) throws Exception {
                    List<ResultItem> resultItems = convertToResultItems(aItems, bItems);    // long process here, convert aItems and bItems to resultItems
                    return resultItems;
                }
            }
    )
            .onBackpressureLatest()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new ResourceSubscriber<List<ResultItem>>() {
                @Override
                public void onNext(List<ResultItem> resultItems) {
                    adapter.addData(resultItems);
                }

                @Override
                public void onError(Throwable t) {
                }

                @Override
                public void onComplete() {
                }
            });
    mSubscriptions.add(subscriber);
}

问题：如果 BiFunction 运行时间过长（例如 10 秒），比 observables 触发间隔长（例如 observables 每隔 1 秒触发一次），这将导致 BiFunction 做不必要的工作，因为我只需要最新发出的项目，但 BiFunction是一个一个处理发射的项目，所以BiFunction会处理旧的发射项目，我不需要处理它。我希望 BiFunction 在 BiFunction 中每个完成的 apply() 都跳过旧的发射项目并处理最新发射的项目，以减少资源浪费并节省时间。rxjava 是否有类似于 BiFunction 的背压的方法或其他解决此问题的方法？

该图显示了当前和预期的 BiFunction 时间线。图链接

我找到了两种方法来解决这个问题，但是有缺陷。

Method1：通过Pair将“aItems”和“bItems”组合起来，然后将引用传递给switchMap并处理job。

缺陷：switchMap 只向订阅者发送最新的项目，但仍然做不必要的工作。

方法2 ：也结合“aItems”和“bItems”然后将引用传递给onNext并处理作业。

缺陷：阻塞了 UI 线程。

4

1 回答 1

0

combineLatest您可以只传递' 的组合器函数中的这对值，并用于observeOn将计算置于原始源线程之外：

 .combineLatest(srcA, srcB, (a, b) -> Pair.of(a, b))
 .onBackpressureLatest()
 .observeOn(Schedulers.computation(), false, 1)
 .map(pair -> compute(pair))
 .observeOn(AndroidSchedulers.mainThread())
 ...
于 2017-06-19T06:54:09.487 回答