Rxjava的一種用途是他可以統一所以回撥介面。

現有個資料庫請求的回撥，類似下邊：

  @Override
    public void getUsers(LoadUsersCallback callback) {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                final List<User> users = mCoffeeDao.getUsers();
                mAppExecutors.mainThread().execute(new Runnable() {
                    @Override
                    public void run() {
                        if (users.isEmpty()){
                            //沒有使用者
                            callback.onDataNotAvailable();
                        }else {
                            callback.onUsersLoaded(users);
                        }
                    }
                });
            }
        };

        mAppExecutors.singleThread().execute(runnable);
    }
 

將這種介面改裝成Rx程式設計模式

    @Override
    public Flowable<List<User>> getUsers() {                      
        return Flowable.create(new FlowableOnSubscribe<List<User>>() {
            @Override
            public void subscribe(FlowableEmitter<List<User>> emitter) throws Exception {
               // mCoffeeDao.getUsers();
                final List<User> users = mCoffeeDao.getUsers();
                if (!users.isEmpty()) {
                    emitter.onNext(users);
                }else {
                    emitter.onError(new Throwable("user庫沒有資料"));
                }
                emitter.onComplete();

            }
        },BackpressureStrategy.BUFFER);
    }
 

然後呼叫的時候：

mCompositeDisposable.clear();
        Disposable disposable = localDataSourceRx.getUsers()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<List<User>>() {
                    @Override
                    public void accept(List<User> users) throws Exception {
                        if (users != null) {
                            for (User per : users) {
                                MyLog.i(TAG, per.toString());
                            }
                        }
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        MyLog.i(TAG, throwable.getMessage());
                    }
                }, () -> {
                    MyLog.i(TAG, "完成: ");
                });
        mCompositeDisposable.add(disposable);