RxJava2.0中map操作符用法和源码分析

Posted 2022-12-13 yuminfeng728

map基本使用

map是变换操作符，对原始Observable发射的每一项数据应用一个你选择的函数生成新的结果，然后返回一个发射这些结果Observable。

但从字面上还是比较难以理解，我们可以用代码示例说明：

Observable.just(1,2,3).map(new Function<Integer, String>() 
    @Override
    public String apply(Integer integer) throws Exception 
        return "This is new result " + integer;
    
).subscribe(new Consumer<String>() 
    @Override
    public void accept(String s) throws Exception 
        println("accept : " + s +"\\n");
    
);

输出结果：
accept : This is new result 1

accept : This is new result 2

accept : This is new result 3

由上面代码可知，执行map操作时，首先接收原始Observable发射的数据，然后根据你的操作生成新的数据并将这些新的数据发射，这时观察者中接收的就是新生成的数据。

下面我们将从源码的角度来分析下：

这里我们首先使用just操作符创建一个Observable来发射指定的数据。关于just如何创建Observable对象，我们这里不做分析，前面文章中已经说明。这里just创建的具体对象为ObservableFromArray。我们直接分析map的源码，我们看到在调用map方法时，我们需要传入一个Function的对象：

/**
 * A functional interface that takes a value and returns another value, possibly with a
 * different type and allows throwing a checked exception.
 *
 * @param <T> the input value type
 * @param <R> the output value type
 */
public interface Function<T, R> 
    /**
     * Apply some calculation to the input value and return some other value.
     * @param t the input value
     * @return the output value
     * @throws Exception on error
     */
    @NonNull
    R apply(@NonNull T t) throws Exception;

如上所述，这个接口的功能主要是接收一个值(T)，然后返回另一个值(R)。

我们在查看map的方法：

@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));

与之前其它操作符一样的调用逻辑，将当前的Observable对象和生成的Function对象作为参数，生成一个ObservableMap的对象。

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> 
    final Function<? super T, ? extends U> function;

    public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) 
        super(source);
        this.function = function;
    

    @Override
    public void subscribeActual(Observer<? super U> t) 
        source.subscribe(new MapObserver<T, U>(t, function));
    

    ......

完成了Observable对象初始化后，我们开始订阅观察者。这里我们选择使用的观察者为Consumer对象。订阅观察者时，执行subscribe方法：
Observable#subscribe

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable subscribe(Consumer<? super T> onNext) 
    return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());



@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
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方法中执行subscribeActual(observer)方法:

@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");

        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方法中，首先会创建一个MapObserver对象，参数t对应的是LambdaObserver。
然后执行source.subscribe方法，source代表的是之前的Observable对象，也就是just创建的ObservableFromArray对象，所以再次调用Observable中subscribe方法，执行subscribeActual，而这次执行的对象是ObservableFromArray，而参数observer具体实现是MapObserver的对象：

public final class ObservableFromArray<T> extends Observable<T> 
    final T[] array;
    public ObservableFromArray(T[] array) 
        this.array = array;
    
    @Override
    public void subscribeActual(Observer<? super T> s) 
        FromArrayDisposable<T> d = new FromArrayDisposable<T>(s, array);

        s.onSubscribe(d);

        if (d.fusionMode) 
            return;
        

        d.run();
    

    ......    

到这里执行的逻辑与之前分析fromArray操作符用法相同，不再做具体分析。在FromArrayDisposable中具体执行的run方法中：
FromArrayDisposable#run

void run() 
    T[] a = array;
    int n = a.length;

    for (int i = 0; i < n && !isDisposed(); i++) 
        T value = a[i];
        if (value == null) 
            actual.onError(new NullPointerException("The " + i + "th element is null"));
            return;
        
        actual.onNext(value);
    
    if (!isDisposed()) 
        actual.onComplete();
    

如上此时actual是由MapObserver实现的，我们看下MapObserver的onNext方法：
MapObserver#onNext：

@Override
public void onNext(T t) 
    if (done) 
        return;
    

    if (sourceMode != NONE) 
        actual.onNext(null);
        return;
    

    U v;

    try 
        v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
     catch (Throwable ex) 
        fail(ex);
        return;
    
    actual.onNext(v);

当调用onNext方法时，通过Function接口回调apply方法获得转换后的数据，然后再通过 actual.onNext(v)方法发射出去。此时的actual中的onNext方法就可以接收新的参数了，而actual就是之前初始化的LambdaObserver对象。通过它可以让Consumer的accept方法中接收该数据。这里的执行逻辑之前已经分析过，这里不再详述。