正确使用RxJava2中的单控制流

implementation 'io.reactivex.rxjava2:rxjava:2.x.x'

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        // 在这里定义数据流的发射逻辑
        emitter.onNext(1);
        emitter.onNext(2);
        emitter.onNext(3);
        emitter.onComplete();
    }
});

Observer<Integer> observer = new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        // 在这里进行订阅时的初始化操作
    }

    @Override
    public void onNext(Integer value) {
        // 在这里处理接收到的数据
    }

    @Override
    public void onError(Throwable e) {
        // 在这里处理发生的错误
    }

    @Override
    public void onComplete() {
        // 在这里处理数据流结束的逻辑
    }
};

observable.subscribe(observer);

observable.map(new Function<Integer, String>() {
    @Override
    public String apply(Integer value) throws Exception {
        return "转换后的值：" + value;
    }
}).subscribe(observer);

observable.subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(observer);

observable.onErrorResumeNext(new Function<Throwable, ObservableSource<Integer>>() {
    @Override
    public ObservableSource<Integer> apply(Throwable throwable) throws Exception {
        // 返回备用的数据流
        return Observable.just(4, 5, 6);
    }
}).subscribe(observer);

Disposable disposable = observable.subscribe(observer);
// 在需要取消订阅时调用dispose()方法
disposable.dispose();