使用PublishSubject发布给特定观察者

基本上可以定义两种不同的方法来获取 Observable PublishSubject . 第一个observable将发射所有项目,第二个仅发射流行项目:

public class DataStore {

    private PublishSubject<DataItem> dataItemPublishSubject = PublishSubject.create();

    public Flowable<DataItem> getAllObservable() {
        return dataItemPublishSubject.toFlowable(BackpressureStrategy.BUFFER);
    }

    public Flowable<DataItem> getPopularObservable() {
        return dataItemPublishSubject.toFlowable(BackpressureStrategy.BUFFER)
                .filter(new Predicate<DataItem>() {
                    @Override
                    public boolean test(DataItem dataItem) throws Exception {
                        return dataItem.popular;
                    }
                });
    }

    public static class DataItem {
        public final boolean popular;

        public DataItem(boolean popular) {
            this.popular = popular;
        }
    }
}

如果你不想使用两种方法,你可以移动 .filter()

    dataStore.getAllObservable()
            .doOnNext(new Consumer<DataStore.DataItem>() {
                @Override
                public void accept(DataStore.DataItem dataItem) throws Exception {
                    pagerAdapter.addDataAll(dataItem);      
                }
            })
            .filter(new Predicate<DataStore.DataItem>() {
                @Override
                public boolean test(DataStore.DataItem dataItem) throws Exception {
                    return dataItem.popular;
                }
            })
            .doOnNext(new Consumer<DataStore.DataItem>() {
                @Override
                public void accept(DataStore.DataItem dataItem) throws Exception {
                    pagerAdapter.addDataPopular(dataItem);
                }
            })
            .subscribe();