RxJava 源码剖析：以 Create 操作符一窥创建型操作符的奥秘

2023-10-23 17:33:08

揭秘 RxJava Create 操作符：打造可观测序列的基石

初探 Create 操作符

RxJava 是一个强大的响应式编程框架，Create 操作符是其核心创建型操作符之一。它赋予开发者自定义可观测序列的能力，使其能够在特定的场景下灵活地处理事件。

剖析 Create 操作符的源码

Create 操作符的源码位于 RxJava 库的 Observable.create() 方法中。它接受一个 ObservableOnSubscribe 实例作为参数，后者定义了可观测序列发出的事件逻辑：

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    return new Observable<>(source);
}

由此可见，Create 操作符通过将 ObservableOnSubscribe 实例封装在 Observable 对象中来创建可观测序列。

深入理解 ObservableOnSubscribe

ObservableOnSubscribe 是一个接口，负责定义可观测序列发出事件的逻辑。其唯一的 subscribe() 方法接收一个 Observer 实例，用于处理事件：

public interface ObservableOnSubscribe<T> {
    void subscribe(Observer<? super T> observer) throws Exception;
}

Create 操作符通过 ObservableOnSubscribe.subscribe() 方法将可观测序列与观察者连接起来，使可观测序列能够向观察者发出事件。

示例详解

以下代码演示了 Create 操作符的简单使用：

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(Observer<? super Integer> observer) throws Exception {
        observer.onNext(1);
        observer.onNext(2);
        observer.onNext(3);
        observer.onComplete();
    }
})
.subscribe(new Observer<Integer>() {
    @Override
    public void onNext(Integer integer) {
        System.out.println(integer);
    }

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

    @Override
    public void onComplete() {
        System.out.println("Complete!");
    }
});