RxJava源码解析03-变换过程(map)

柔光的暖阳◎ 2023-10-17 15:33 29阅读 0赞

### RxJava源码解析03-变换过程(map) ###

#### map操作符 ####

将一个Observable转化为一个新的Observable对象并发射。

Observable.just(123)
        .map(new Func1<Integer, String>() {
            @Override
            public String call(Integer integer) {
                return integer + "abc";
            }
        })
        .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            System.out.println(s);
        }
    });

来分析下这个变换过程

public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
        return unsafeCreate(new OnSubscribeMap<T, R>(this, func));
    }

先来看OnSubscribeMap的构造方法

public OnSubscribeMap(Observable<T> source, Func1<? super T, ? extends R> transformer) {
        this.source = source;
        this.transformer = transformer;
    }

可以看到，这里将Observable和Func1分别赋值给了OnSubscribeMap的source和transformer。

再来看unsafeCreate()

public static <T> Observable<T> unsafeCreate(OnSubscribe<T> f) {
        return new Observable<T>(RxJavaHooks.onCreate(f));
    }

可以看到这里返回了一个Observable对象(观察者)，而RxJavaHooks.onCreate()其实就是返回f对象(也就OnSubscribeMap，是被观察者)。  
所以，map()最终返回了一个新的Observable对象，而其onSubscribe(被观察者)是OnSubscribeMap。  
之后，和订阅过程就没有什么区别了，最终会调用OnSubscribeMap的call方法

public final class OnSubscribeMap<T, R> implements OnSubscribe<R> {
        final Observable<T> source;
    
        final Func1<? super T, ? extends R> transformer;
    
        public OnSubscribeMap(Observable<T> source, Func1<? super T, ? extends R> transformer) {
            this.source = source;
            this.transformer = transformer;
        }
    
        @Override
        public void call(final Subscriber<? super R> o) {
            MapSubscriber<T, R> parent = new MapSubscriber<T, R>(o, transformer);
            o.add(parent);
            source.unsafeSubscribe(parent);
        }
    }

可以看到，OnSubscribeMap的call()中new 了一个MapSubscriber，并把transformer传入，并调用了source.unsafeSubscribe()执行订阅。

static final class MapSubscriber<T, R> extends Subscriber<T> {
    
        final Subscriber<? super R> actual;
    
        final Func1<? super T, ? extends R> mapper;
    
        boolean done;
    
        public MapSubscriber(Subscriber<? super R> actual, Func1<? super T, ? extends R> mapper) {
            this.actual = actual;
            this.mapper = mapper;
        }
    
        @Override
        public void onNext(T t) {
            R result;
    
            try {
                result = mapper.call(t);
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                unsubscribe();
                onError(OnErrorThrowable.addValueAsLastCause(ex, t));
                return;
            }
    
            actual.onNext(result);
        }
    
        @Override
        public void onError(Throwable e) {
            if (done) {
                RxJavaHooks.onError(e);
                return;
            }
            done = true;
    
            actual.onError(e);
        }
    
    
        @Override
        public void onCompleted() {
            if (done) {
                return;
            }
            actual.onCompleted();
        }
    
        @Override
        public void setProducer(Producer p) {
            actual.setProducer(p);
        }
    }

MapSubscriber中，mapper就是`map()`传入参数的那个Func1，actual就是`call()`传入参数的那个Subscriber。

MapSubscriber中主要来看onNext()，首先会执行mapper.call(t)，执行返回结果之后，再交给actual.onNext()进行调用。

#### 总结 ####

RxJava map的变换过程其实就是在原来OnSubscribe的基础上再包装一层生成OnSubscribeMap(代理模式)，在这个代理的`OnNext()`中，优先执行`map(Func1 func)`中的func方法，从而实现参数类型的变化，执行完毕后，再将返回值传入，执行Subscriber的`OnNext()`