RxJava的部分变换操做符源码分析

一. map操做符

（一）定义

对Observable发射的每一项数据应用一个函数，执行变换操做，以下图。

map操做符对原始Observable发射的的每一项数据应用一个你选择的函数，而后返回一个发射这些结果的Observable。

RxJava将这个操做符实现为map函数，这个操做符默认再也不任何特定的调度器上执行。

（二）示例

public void mapTest(){
        Observable.just("HELLO")
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        return s.toLowerCase();
                    }
                })
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        return s + " world";
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println(s);
                    }
                });
    }
复制代码

输出结果：

hello world
复制代码

（三）源码分析

1.调用map()方法

/** * Returns an Observable that applies a specified function to each item emitted by the source ObservableSource and * emits the results of these function applications. * * @param <R> the output type * @param mapper * a function to apply to each item emitted by the ObservableSource * @return an Observable that emits the items from the source ObservableSource, transformed by the specified * function */
    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
    }
复制代码

2.底层在调用ObservableMap类

将自定义的Function对象赋值给成员变量：function. 该类中一样有个重写的subscribeActual()方法， 在

@Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MapObserver<T, U>(t, function));
    }
复制代码

后续调用subscribe()时，其会先调用Observable类中的subscribe()方法

/**
     * Subscribes to an ObservableSource and provides a callback to handle the items it emits.
     *
     * @param onNext
     *             the {@code Consumer<T>} you have designed to accept emissions from the ObservableSource
     * @return a {@link Disposable} reference with which the caller can stop receiving items before
     *         the ObservableSource has finished sending them
     * @throws NullPointerException
     *             if {@code onNext} is null
     */
    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(Consumer<? super T> onNext) {
        return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }
复制代码

subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer())该方法会调用Observable同类中的重载的方法：subscribe()

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {
        ObjectHelper.requireNonNull(onNext, "onNext is null");
        ObjectHelper.requireNonNull(onError, "onError is null");
        ObjectHelper.requireNonNull(onComplete, "onComplete is null");
        ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");

        LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
        
        // 核心方法
        subscribe(ls);

        return ls;
    }
复制代码

该方法又会调用同类中重载的subscribe()方法

@SchedulerSupport(SchedulerSupport.NONE)
    @Override
    public final void subscribe(Observer<? super T> observer) {
        ObjectHelper.requireNonNull(observer, "observer is null");
        try {
            observer = RxJavaPlugins.onSubscribe(this, observer);

            ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
            
            // 核心方法，继承Observable的类都须该方法
            subscribeActual(observer);
            
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            // can't call onError because no way to know if a Disposable has been set or not
            // can't call onSubscribe because the call might have set a Subscription already
            RxJavaPlugins.onError(e);

            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
            npe.initCause(e);
            throw npe;
        }
    }
复制代码

此时就会调用ObservableMap类中的subscribeActual()方法实现，

该方法又会调用上述方法，不过此时subscribeActual()方法是ObservableJust类中的方法

@Override
    protected void subscribeActual(Observer<? super T> s) {
        ScalarDisposable<T> sd = new ScalarDisposable<T>(s, value);
        s.onSubscribe(sd);
        sd.run();
    }
复制代码

该方法就会使用s.onSubscribe()来BasicFuseableObserver类中的OnSubscribe()方法

// final: fixed protocol steps to support fuseable and non-fuseable upstream
    @SuppressWarnings("unchecked")
    @Override
    public final void onSubscribe(Disposable s) {
        if (DisposableHelper.validate(this.s, s)) {

            this.s = s;
            if (s instanceof QueueDisposable) {
                this.qs = (QueueDisposable<T>)s;
            }

            if (beforeDownstream()) {
            
                //对应的代码
                actual.onSubscribe(this);

                afterDownstream();
            }

        }
    }
复制代码

进过屡次的调用，最终就会调用咱们再主方法中重写的accept()方法，输出对应的结果。

二.flatmap操做符

（一）定义

flatMap将一个发射数据的Observable变换为多个Observable，而后将它们发射的数据合并放进一个单独的Observable，如图2：

flapMap操做符使用一个指定的函数对原始Observable发射的每一项数据执行变换操做，这个函数返回一个自己也发射数据的Observable，而后flatMap合并这些Observables发射的数据，最后将合并后的结果看成它本身的数据序列发射。

（二）示例

数据类

public class User {
    public String userName;
    public List<Address> addresses;

    public static class Address {
        public String street;
        public String city;
    }
}
复制代码

public void flatMapTest() {
        User user = new User();
        user.userName = "tony";
        user.addresses = new ArrayList<>();
        User.Address address1 = new User.Address();
        address1.street = "ren ming road";
        address1.city = "Su zhou";
        user.addresses.add(address1);

        User.Address address2 = new User.Address();
        address2.street = "dong wu bei road";
        address2.city = "Su zhou";
        user.addresses.add(address2);

        Observable.just(user)
                .flatMap(new Function<User, ObservableSource<User.Address>>() {
                    @Override
                    public ObservableSource<User.Address> apply(User user) throws Exception {
                        return Observable.fromIterable(user.addresses);
                    }
                })
                .subscribe(new Consumer<User.Address>() {
                    @Override
                    public void accept(User.Address address) throws Exception {
                        System.out.println(address.street);
                    }
                });
    }
复制代码

输出结果：

ren ming road
dong wu bei road
复制代码

(三) 源码分析

1.flatMap()

flatMap()底层会调用有多个重载的方法，最终会调用以下方法：

/** * Returns an Observable that emits items based on applying a function that you supply to each item emitted * by the source ObservableSource, where that function returns an ObservableSource, and then merging those resulting * ObservableSources and emitting the results of this merger, while limiting the maximum number of concurrent * subscriptions to these ObservableSources. * * @param <R> the value type of the inner ObservableSources and the output type * @param mapper * a function that, when applied to an item emitted by the source ObservableSource, returns an * ObservableSource * @param maxConcurrency * the maximum number of ObservableSources that may be subscribed to concurrently * @param delayErrors * if true, exceptions from the current Observable and all inner ObservableSources are delayed until all of them terminate * if false, the first one signalling an exception will terminate the whole sequence immediately * @param bufferSize * the number of elements to prefetch from each inner ObservableSource * @return an Observable that emits the result of applying the transformation function to each item emitted * by the source ObservableSource and merging the results of the ObservableSources obtained from this * transformation */
    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper, boolean delayErrors, int maxConcurrency, int bufferSize) {
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
        
        if (this instanceof ScalarCallable) {
            @SuppressWarnings("unchecked")
            T v = ((ScalarCallable<T>)this).call();
            if (v == null) {
                return empty();
            }
            return ObservableScalarXMap.scalarXMap(v, mapper);
        }
        return RxJavaPlugins.onAssembly(new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize));
    }
复制代码

然后根据this是否为ScalarCallable类，返回不一样的对象：

本示例是该示例，进而调用ObservableScalarXMap.scalarXMap(v, mapper);

/** * Maps a scalar value into an Observable and emits its values. * * @param <T> the scalar value type * @param <U> the output value type * @param value the scalar value to map * @param mapper the function that gets the scalar value and should return * an ObservableSource that gets streamed * @return the new Observable instance */
    public static <T, U> Observable<U> scalarXMap(T value, Function<? super T, ? extends ObservableSource<? extends U>> mapper) {
        return RxJavaPlugins.onAssembly(new ScalarXMapObservable<T, U>(value, mapper));
    }
复制代码

进而建立ScalarXMapObservable对象。

2.subscribe()

该方法也有多个重载方法，最终会调用subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe)并执行其中的**subscribe(ls);方法 进而再调用subscribe(Observer<? super T> observer)并执行其中的subscribeActual(observer);**方法

进而调用ObservableScalarXMap的**subscribeActual(Observer<? super R> s)**方法

@SuppressWarnings("unchecked")
        @Override
        public void subscribeActual(Observer<? super R> s) {
            ObservableSource<? extends R> other;
            try {
                // 核心代码
                // 这儿调用在用户自定义的flatMap方法中自定义apply()方法
                other = ObjectHelper.requireNonNull(mapper.apply(value), "The mapper returned a null ObservableSource");
            } catch (Throwable e) {
                EmptyDisposable.error(e, s);
                return;
            }
            
            // 此处是判断other的类型，进而执行不一样的方法
            if (other instanceof Callable) {
                R u;

                try {
                    u = ((Callable<R>)other).call();
                } catch (Throwable ex) {
                    Exceptions.throwIfFatal(ex);
                    EmptyDisposable.error(ex, s);
                    return;
                }

                if (u == null) {
                    EmptyDisposable.complete(s);
                    return;
                }
                ScalarDisposable<R> sd = new ScalarDisposable<R>(s, u);
                s.onSubscribe(sd);
                sd.run();
            } else {
                // 由于other是Observable类型，于是执行这步
                other.subscribe(s);
            }
        }
    }
复制代码

在subscribeActual()调用mapper.apply(value)执行用户自定义的flatMap方法中的apply()方法。

此时又会调用Observable类的subscribe()方法。

注意此处的subscribe()和上面已经调用的Observable不是同一个对象！！！

再次调用该方法中的subscribeActual()方法，进而调用ObservableFromIterable类中的**subscribeActual()**方法：

@Override
    public void subscribeActual(Observer<? super T> s) {
        Iterator<? extends T> it;
        try {
            // 此处对source的数据进行依次发送
            it = source.iterator();
            
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            EmptyDisposable.error(e, s);
            return;
        }
        boolean hasNext;
        try {
            // 判断是否it ------------------------------------------
            hasNext = it.hasNext();
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            EmptyDisposable.error(e, s);
            return;
        }
        if (!hasNext) {
            EmptyDisposable.complete(s);
            return;
        }

        FromIterableDisposable<T> d = new FromIterableDisposable<T>(s, it);
        s.onSubscribe(d);

        if (!d.fusionMode) {
            d.run();
        }
    }
复制代码

(四)总结