Rxjava2 常见使用场景

1.执行倒计时操作

• 使用操作符 interval
  • 第一个参数（initialDelay）：延迟执行时间
  • 第二个参数 （period）：发射时间间隔
  • 第三个参数 （unit）：时间单位

  private void startCountdown(int countdownTime) {
        Observable.interval(0, 1, TimeUnit.SECONDS).take(countdownTime).map(new Function<Long, Long>() {
            @Override
            public Long apply(Long aLong) throws Exception {
                //强制格式化一次剩余时间（延时 mTime 秒 最后倒计时结束显示 0 则需要多 减去 1）
                return countdownTime - 1 - aLong;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        mTvTime.setText(String.valueOf(aLong));
                    }
                });

    }

如果想监听倒计时完成和异常等

       Observable.interval(0, 1, TimeUnit.SECONDS).take(countdownTime).map(new Function<Long, Long>() {
            @Override
            public Long apply(Long aLong) throws Exception {        
                return countdownTime - 1 - aLong;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {                
                    }
                    @Override
                    public void onNext(Long aLong) {
                        mTvTime.setText(String.valueOf(aLong));
                    }
                    @Override
                    public void onError(Throwable e) {
                    }
                    @Override
                    public void onComplete() {
                    }
                });

2.线程执行耗时操作后，返回主线程执行结果

 private CompositeDisposable mCompositeDisposable = new CompositeDisposable();
 Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 100; i++) {
                    if (i % 20 == 0) {
                        try {
                            Thread.sleep(500); //模拟耗时的操作。
                        } catch (InterruptedException exception) {
                            if (!e.isDisposed()) {
                                e.onError(exception);
                            }
                        }
                        e.onNext(i);
                    }
                }
                e.onComplete();
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        mCompositeDisposable.add(d);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        mTvBa.setText(String.valueOf(integer));
                    }

                    @Override
                    public void onError(Throwable e) {
                        mTvBa.setText("error");
                    }

                    @Override
                    public void onComplete() {
                        mTvBa.setText("complete");
                    }
                });
 @Override
    protected void onDestroy() {
        super.onDestroy();
        mCompositeDisposable.clear();
    }