在ReactiveX中为另一个主题创建计算主题

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;

Observable<Integer> sourceObservable = Observable.just(1, 2, 3, 4, 5);

Observable<Integer> computedObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        sourceObservable.subscribe(
            // 处理源主题的事件序列
            new Consumer<Integer>() {
                @Override
                public void accept(Integer value) throws Exception {
                    // 进行计算和转换
                    int computedValue = value * 2;
                    // 发射计算结果
                    emitter.onNext(computedValue);
                }
            },
            // 处理源主题的错误事件
            new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    // 处理错误
                    emitter.onError(throwable);
                }
            },
            // 处理源主题的完成事件
            new Action() {
                @Override
                public void run() throws Exception {
                    // 结束事件序列
                    emitter.onComplete();
                }
            }
        );
    }
});

computedObservable.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer value) throws Exception {
        // 处理计算主题的事件序列
        System.out.println("Computed value: " + value);
    }
});