RxJava 源码浅析：揭秘订阅流程和 Map/Filter 操作符的实现

2023-11-04 09:50:40

RxJava 源码浅析——订阅流程、Map 与 Filter 操作符实现原理

前言

RxJava 是一个基于观察者模式的响应式编程框架，在移动开发和服务器开发领域应用广泛。掌握 RxJava 的实现原理有助于我们深入理解其设计思想，从而更熟练地使用它编写高效代码。本文将带你深入探究 RxJava 源码，剖析订阅流程以及 Map 和 Filter 操作符的实现原理。

订阅流程

1. 创建 Observable

RxJava 中，数据源被抽象为 Observable，它负责发出事件（数据）。我们可以通过各种方式创建 Observable，例如：

Observable<Integer> numbers = Observable.just(1, 2, 3);

2. 创建 Observer

Observer 负责接收 Observable 发出的事件。我们可以通过实现 Observer 接口或使用 RxJava 提供的便捷方法创建 Observer：

Observer<Integer> observer = new Observer<Integer>() {
    @Override
    public void onNext(Integer integer) {
        System.out.println(integer);
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("Error: " + e.getMessage());
    }

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

3. 订阅

订阅操作将 Observable 和 Observer 关联起来。通过调用 Observable 的 subscribe() 方法可以实现订阅：

numbers.subscribe(observer);

4. 事件发送

订阅后，Observable 会依次发出事件，包括 onNext()（数据事件）、onError()（错误事件）和 onComplete()（完成事件）。

5. 回调处理

Observer 会根据收到的事件类型回调相应的方法。例如，当收到数据事件时，onNext() 方法会被调用。

Map 操作符

Map 操作符用于将 Observable 中的每一项数据转换成为另一项数据。它的实现原理如下：

1. 创建 MapTransformer

RxJava 内部类 MapTransformer 实现了 ObservableOperator 接口，负责处理 Map 操作符的转换逻辑：

private static final class MapTransformer<T, R> implements ObservableOperator<R, T> {
    private final Function<? super T, ? extends R> transformer;

    public MapTransformer(Function<? super T, ? extends R> transformer) {
        this.transformer = transformer;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super R> subscriber) {
        return new MapSubscriber<>(subscriber, transformer);
    }
}

2. 创建 MapSubscriber

MapSubscriber 实现了 Subscriber 接口，负责接收 Observable 发出的事件并进行转换：

private static final class MapSubscriber<T, R> extends Subscriber<T> {
    private final Subscriber<? super R> actual;
    private final Function<? super T, ? extends R> transformer;

    public MapSubscriber(Subscriber<? super R> actual, Function<? super T, ? extends R> transformer) {
        this.actual = actual;
        this.transformer = transformer;
    }

    @Override
    public void onNext(T t) {
        R mappedValue;
        try {
            mappedValue = transformer.apply(t);
        } catch (Throwable e) {
            actual.onError(e);
            return;
        }
        actual.onNext(mappedValue);
    }

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

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

3. 订阅处理

当 Observable 调用 Map 操作符时，MapTransformer 会被创建，并调用 call() 方法创建一个 MapSubscriber。MapSubscriber 订阅 Observable，并对接收到的数据项进行转换。转换后的数据项被发送给下游的 Observer。

Filter 操作符

Filter 操作符用于过滤 Observable 发出的数据项，仅允许满足指定条件的数据项通过。它的实现原理如下：

1. 创建 FilterTransformer

RxJava 内部类 FilterTransformer 实现了 ObservableOperator 接口，负责处理 Filter 操作符的过滤逻辑：

private static final class FilterTransformer<T> implements ObservableOperator<T, T> {
    private final Predicate<? super T> predicate;

    public FilterTransformer(Predicate<? super T> predicate) {
        this.predicate = predicate;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super T> subscriber) {
        return new FilterSubscriber<>(subscriber, predicate);
    }
}

2. 创建 FilterSubscriber

FilterSubscriber 实现了 Subscriber 接口，负责接收 Observable 发出的事件并进行过滤：

private static final class FilterSubscriber<T> extends Subscriber<T> {
    private final Subscriber<? super T> actual;
    private final Predicate<? super T> predicate;

    public FilterSubscriber(Subscriber<? super T> actual, Predicate<? super T> predicate) {
        this.actual = actual;
        this.predicate = predicate;
    }

    @Override
    public void onNext(T t) {
        boolean keep;
        try {
            keep = predicate.test(t);
        } catch (Throwable e) {
            actual.onError(e);
            return;
        }
        if (keep) {
            actual.onNext(t);
        }
    }

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

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

3. 订阅处理

当 Observable 调用 Filter 操作符时，FilterTransformer 会被创建，并调用 call() 方法创建一个 FilterSubscriber。FilterSubscriber 订阅 Observable，并对接收到的数据项进行过滤。仅满足条件的数据项会被发送给下游的 Observer。