RxJava2源码大揭秘：剥茧抽丝，洞察精髓

2024-02-14 03:22:10

RxJava2源码初探

让我们从一个简单的例子开始，我们先创建一个Observable对象，然后使用subscribe()方法来订阅它，最后使用onNext()方法来接收它的事件。

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

observable.subscribe(new Observer<Integer>() {
    @Override
    public void onNext(Integer item) {
        System.out.println(item);
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
});

运行这段代码，你会看到控制台输出1, 2, 3, 4, 5。

现在我们来看看RxJava2的源码，看看它是如何实现的。

首先，我们来看看Observable类的just()方法。这个方法的作用是创建一个包含指定元素的Observable对象。

public static <T> Observable<T> just(T... items) {
    return fromIterable(Arrays.asList(items));
}

just()方法内部调用了fromIterable()方法。这个方法的作用是创建一个包含指定集合的Observable对象。

public static <T> Observable<T> fromIterable(Iterable<? extends T> iterable) {
    return create(new ObservableFromIterable<>(iterable));
}

fromIterable()方法内部调用了create()方法。这个方法的作用是创建一个新的Observable对象。

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

create()方法内部创建了一个新的Observable对象，并将ObservableOnSubscribe对象作为参数传递给它。

ObservableOnSubscribe对象是一个接口，它定义了subscribe()方法。subscribe()方法的作用是订阅Observable对象，并将Observer对象作为参数传递给它。

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

ObservableFromIterable类实现了ObservableOnSubscribe接口。它的subscribe()方法内部遍历了指定的集合，并将每个元素作为参数调用Observer对象的onNext()方法。

public final class ObservableFromIterable<T> implements ObservableOnSubscribe<T> {

    private final Iterable<? extends T> iterable;

    public ObservableFromIterable(Iterable<? extends T> iterable) {
        this.iterable = iterable;
    }

    @Override
    public void subscribe(Observer<? super T> observer) {
        try {
            for (T item : iterable) {
                observer.onNext(item);
            }
            observer.onComplete();
        } catch (Throwable e) {
            observer.onError(e);
        }
    }
}