rxjava怎样实现自定义操作符

在 RxJava 中，你可以通过创建一个新的 ObservableTransformer 或者使用 Observable.create() 方法来实现自定义操作符。下面是两种方法的示例：

方法1：使用 ObservableTransformer

import io.reactivex.Observable;
import io.reactivex.ObservableTransformer;

public class CustomOperatorExample {

    public static void main(String[] args) {
        Observable.just(1, 2, 3, 4, 5)
                .compose(customOperator())
                .subscribe(System.out::println);
    }

    public static <T> ObservableTransformer<T, T> customOperator() {
        return upstream -> upstream.map(value -> value * 2);
    }
}

在这个例子中，我们创建了一个名为 customOperator 的方法，它接受一个 Observable<T> 类型的参数，并返回一个 ObservableTransformer<T, T> 类型的对象。在这个方法中，我们使用 map 操作符将每个值乘以2。

方法2：使用 Observable.create()

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

public class CustomOperatorExample {

    public static void main(String[] args) {
        Observable.just(1, 2, 3, 4, 5)
                .lift(customOperator())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        // Handle subscription
                    }

                    @Override
                    public void onNext(Integer value) {
                        System.out.println(value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        // Handle error
                    }

                    @Override
                    public void onComplete() {
                        // Handle completion
                    }
                });
    }

    public static ObservableOperator<Integer, Integer> customOperator() {
        return subscriber -> new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                subscriber.onSubscribe(d);
            }

            @Override
            public void onNext(Integer value) {
                subscriber.onNext(value * 2);
            }

            @Override
            public void onError(Throwable e) {
                subscriber.onError(e);
            }

            @Override
            public void onComplete() {
                subscriber.onComplete();
            }
        };
    }
}

在这个例子中，我们创建了一个名为 customOperator 的方法，它返回一个 ObservableOperator<Integer, Integer> 类型的对象。在这个方法中，我们创建了一个新的 Observer<Integer> 对象，并在其 onNext 方法中将每个值乘以2。

这两种方法都可以实现自定义操作符，你可以根据自己的需求选择合适的方法。