Disposable subscription = RxTextView.textChangeEvents(etSearch)
                    .debounce(200, TimeUnit.MILLISECONDS)
                    .observeOn(AndroidSchedulers.mainThread())
                    .toFlowable(BackpressureStrategy.BUFFER)
                    .switchMap(new Function<TextViewTextChangeEvent, Publisher<RealmResults<Note>>>() {
                        @Override
                        public Publisher<RealmResults<Note>> apply(TextViewTextChangeEvent event) throws Exception {
                            String etContent = event.getText().toString();
                            if (etContent.length() > 0) {
                                mLv.setVisibility(View.VISIBLE);
                            } else {
                                mLv.setVisibility(View.GONE);
                            }
                            if (etContent.length() > 0) {
                                mLv.setVisibility(View.VISIBLE);
                            } else {
                                mLv.setVisibility(View.GONE);
                            }
                            finalMAdapter.setSearchStr(etContent);
                            if (dirId == MyApplication.DEFAULT_DIR_ID) {
                                return Realm.getDefaultInstance().where(Note.class)
                                    .contains("title", etContent)
                                    .sort("timeMills", Sort.DESCENDING).findAllAsync().asFlowable();
                            } else {
                                return Realm.getDefaultInstance().where(Note.class)
                                    .equalTo("dirId", dirId)
                                    .contains("title", etContent)
                                    .sort("timeMills", Sort.DESCENDING).findAllAsync().asFlowable();
                            }
                        }
                    })
                    .filter(new Predicate<RealmResults<Note>>() {
                        @Override
                        public boolean test(RealmResults<Note> notes) throws Exception {
                            return notes.isLoaded();
                        }
                    })
                    .subscribe(new Consumer<RealmResults<Note>>() {
                        @Override
                        public void accept(RealmResults<Note> notes) throws Exception {
                            mNotes = notes;
                            if (notes != null && notes.size() > 0) {
                                llNoResult.setVisibility(View.GONE);
                            } else {
                                llNoResult.setVisibility(View.VISIBLE);
                            }
                            finalMAdapter.updateAll(mNotes);
                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            throwable.printStackTrace();
                        }
                    });

private void getAllDatas(final long delay) {
  
	//解除上次订阅
    if (subscribe != null) {
        subscribe.unsubscribe();
        subscribe = null;
    }
  
    subscribe = Observable.timer(delay, TimeUnit.MILLISECONDS)
            .flatMap(new Func1<Long, Observable<GetSomeThingResult>>() {
                @Override
                public Observable<GetSomeThingResult> call(Long aLong) {
                	//接口调用
                    return Api.getDefaultApi().getSomeThing();
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Subscriber<GetSomeThingResult>() {
                @Override
                public void onCompleted() {
  
                }
  
                @Override
                public void onError(Throwable e) {
                    ToastUtils.showCustomBgToast(getString(R.string.no_net_text) + e.toString());
                    getAllDatas(Constants.AUTO_REFRESH_DURATION);
                }
  
                @Override
                public void onNext(GetSomeThingResult result) {
                	//... do some thing with data
  
                    getAllDatas(Constants.AUTO_REFRESH_DURATION);
                }
            });
  
}

//0-100 每隔20ms +1
Observable.just(100)
        .flatMap(new Func1<Integer, Observable<Long>>() {
            @Override
            public Observable<Long> call(Integer integer) {
                return Observable.timer(20, TimeUnit.MILLISECONDS);
            }
        })
        .compose(RxHelper.<Long>io_main())
        .subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
  
            }
  
            @Override
            public void onError(Throwable e) {
  
            }
  
            @Override
            public void onNext(Long aLong) {
                pb.setProgress(aLong.intValue());
            }
        });