一、distinct语法

    public Observable<Integer> getRxJavaDistinctData() {
        return Observable.just(1, 2, 2, 1, 1, 2, 3, 4, 5, 2);
    }
    public void rxJavaDistinctExample() {
        Disposable disposable = model.getRxJavaDistinctData()
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .distinct()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtils.error(TAG, "rxJavaDistinctExample--Consumer--:" 
                        +Thread.currentThread().getName() + "--:" + integer);
                    }
                });
        compositeDisposable.add(disposable);
    }

日志

08-08 17:59:37.074 21405-21405/com.example.zhang E/MainPresenter: rxJavaDistinctExample--Consumer--:main--:1
    rxJavaDistinctExample--Consumer--:main--:2
    rxJavaDistinctExample--Consumer--:main--:3
    rxJavaDistinctExample--Consumer--:main--:4
    rxJavaDistinctExample--Consumer--:main--:5

二、Filter语法

 public void rxJavaFilterExample() {
        Disposable disposable = model.getRxJavaDistinctData()
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        LogUtils.error(TAG, "rxJavaFilterExample--filter--:" + Thread.currentThread().getName() + "--:" + integer);
                        return integer>1;
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtils.error(TAG, "rxJavaFilterExample--Consumer--:" + Thread.currentThread().getName() + "--:" + integer);
                    }
                });
        compositeDisposable.add(disposable);
    }

日志

08-10 08:51:53.009 7170-9373/com.example.zhang E/MainPresenter: rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:1
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:2
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:2
08-10 08:51:53.010 7170-9373/com.example.zhang E/MainPresenter: rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:1
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:1
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:2
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:3
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:4
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:5
    rxJavaFilterExample--filter--:RxCachedThreadScheduler-1--:2
08-10 08:51:53.027 7170-7170/com.example.zhang E/MainPresenter: rxJavaFilterExample--Consumer--:main--:2
    rxJavaFilterExample--Consumer--:main--:2
    rxJavaFilterExample--Consumer--:main--:2
    rxJavaFilterExample--Consumer--:main--:3
    rxJavaFilterExample--Consumer--:main--:4
    rxJavaFilterExample--Consumer--:main--:5
    rxJavaFilterExample--Consumer--:main--:2