引言

前面我们分析了RxJava的线程调度，今天我们研究下RxJava的另外一块强大的功能-事件变换操作符。

map操作符

官方定义：transform the items emitted by an Observable by applying a function to each item

翻译过来就是就是转换发射数据的操作符，说白了就是起到事件变换的作用,下面是图示:

map操作符

map操作符示例:

        /**
         * 变换操作符
         */
        //1.map将事件通过一个函数做变换,一般用作数据类型转换
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
            //Integer转换成String
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "使用 Map变换操作符 将事件" + integer + "的参数从 整型" + integer + " 变换成 字符串类型" + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, s);
            }
        });
        //2.FlatMap/concatMap 将每个事件进行拆分和转换，再合并成一个新的事件序列，最后再发送,前者无序发送，后者有序发送
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                //事件拆分
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("我是事件 " + integer + "拆分后的子事件" + i);
                    // 通过flatMap中将被观察者生产的事件序列先进行拆分，再将每个事件转换为一个新的发送三个String事件
                    // 最终合并，再发送给被观察者
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, s);
            }
        });

在这个例子中map操作符将整形数据变换成了String数据，交给Consumer<String>对象接收。
我们看一下map函数的源码：

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

返回ObservableMap对象:

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    //调用层传入的Function，用于数据转换
    final Function<? super T, ? extends U> function;
    //source为上游，支持类型为T，U为下游的支持类型
    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订阅MapObserver,根据function和传入的观察者t构造MapObserver对象
        source.subscribe(new MapObserver<T, U>(t, function));
    }
    
    static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
       //MapObserver通过function对象mapper，将T数据转换U数据类型后，再转换后的数据交给真正的接受者actual
        final Function<? super T, ? extends U> mapper;

        MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
            super(actual);
            this.mapper = mapper;
        }

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

            if (sourceMode != NONE) {
                actual.onNext(null);
                return;
            }
            //转换后的数据
            U v;

            try {
               //调用mapper.apply(t)执行数据转换，交给actual
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            actual.onNext(v);
        }

        @Override
        public int requestFusion(int mode) {
            return transitiveBoundaryFusion(mode);
        }

        @Override
        public U poll() throws Exception {
            T t = qs.poll();
            return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
        }
    }
}

它继承自AbstractObservableWithUpstream，前面我们提到过，它封装了上游的ObservableSource，是支持数据类型转换的Observable。
subscribeActual订阅方法是调用上游(也就是用户定义的Observable对象)的subscribe方法，入参为Observer对象(下游)的装饰类MapObserver。
ObservableMap持有Function对象和真正的观察者对象，在被订阅的onXXX方法中，通过Function将上游发送的T类型数据转换成U类型数据，然后将交给观察者对象处理。

FlatMap操作符

还是先看看官方定义:

官方定义：transform the items emitted by an Observable by applying a function to each item

大致意思是将每一个上游发射的数据从一个Observable转成为多个Observable，并将所有要发射的数据平铺为一个Observable。
下面是FlatMap的图解:

FlatMap

到这里我们总结一下：

flatmap 转换是一对多的(一对一当然也支持)，原来发射了几个数据，转换之后可以是更多个;
flatMap 转换同样可以改变发射的数据类型;
flatMap 转换后的数据，还是会逐个发射给我们的Observer来接收（就像这些数据是由一个Observable发射的一样，其实是多个Observable发射然后合并的）;
4.注意:转换后的数据发射顺序可能和上游的发射顺序不一致,为什么会这样，我们后面看源码分析。
样例:

 Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                //事件拆分
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("我是事件 " + integer + "拆分后的子事件" + i);
                    // 通过flatMap中将被观察者生产的事件序列先进行拆分，再将每个事件转换为一个新的发送三个String事件
                    // 最终合并，再发送给被观察者
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, s);
            }
        });

下面我们看下源码

FlatMap方法

flatMap有多个重载方法，最终调用下面的方法:

@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);
        }
        //返回一个ObservableFlatMap对象
        return RxJavaPlugins.onAssembly(new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize));
    }

ObservableFlatMap

还是和之前的套路一样，看下ObservableFlatMap的订阅方法:

 @Override
    public void subscribeActual(Observer<? super U> t) {

        if (ObservableScalarXMap.tryScalarXMapSubscribe(source, t, mapper)) {
            return;
        }
       //返回包装类MergeObserver
        source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
    }

最后返回MergeObserver，显而易见这是实现flatMap操作的核心类。

MergeObserver

看关键方法onNext:

  @Override
        public void onNext(T t) {
            // safeguard against misbehaving sources
            if (done) {
                return;
            }
            ObservableSource<? extends U> p;
            try {
               //调用Function方法转换成新的Observable
                p = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null ObservableSource");
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                s.dispose();
                onError(e);
                return;
            }

            if (maxConcurrency != Integer.MAX_VALUE) {
                synchronized (this) {
                    if (wip == maxConcurrency) {
                        sources.offer(p);
                        return;
                    }
                    wip++;
                }
            }
            //默认情况会走到这里
            subscribeInner(p);
        }

由于我们默认调用的flatmap 的 maxConcurrency 大小是 Integer.MAX_VALUE, 所以最终会调用 subscribeInner(p)，注意这里我们的mapper方法以及被调用了，p就是跟我们传入的Function生成的Observable，我们再继续往下看

       @SuppressWarnings("unchecked")
        void subscribeInner(ObservableSource<? extends U> p) {
            for (;;) {
                if (p instanceof Callable) {
                    tryEmitScalar(((Callable<? extends U>)p));

                    if (maxConcurrency != Integer.MAX_VALUE) {
                        synchronized (this) {
                            p = sources.poll();
                            if (p == null) {
                                wip--;
                                break;
                            }
                        }
                    } else {
                        break;
                    }
                } else {
                     //一般情况下，返回的Observable 都不是 Callable类型的，走这里
                    //构造最终的观察者InnerObserver，它是真正接收数据的
                    InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);
                    addInner(inner);
                    p.subscribe(inner);
                    break;
                }
            }
        }

终于找到最后的接受者InnerObserver,接着看它的onNext方法:

        @Override
        public void onNext(U t) {
            if (fusionMode == QueueDisposable.NONE) {
                //MergeObserable的方法
                parent.tryEmit(t, this);
            } else {
                parent.drain();
            }
        }

paren为上面的MergeObserable的tryEmit方法,看字面意思是尝试发送，所以可能会引入同步机制，下面是tryEmit方法:

 void tryEmit(U value, InnerObserver<T, U> inner) {
            //引入同步机制，保证每个时刻只发射一个数据给最后的接受者
            //尝试获取cas锁
            if (get() == 0 && compareAndSet(0, 1)) {
                //最后的接受者
                actual.onNext(value);
                if (decrementAndGet() == 0) {
                    //释放锁返回
                    return;
                }
            } else {//没有拿到锁，就把数据加入队列
                SimpleQueue<U> q = inner.queue;
                if (q == null) {
                    q = new SpscLinkedArrayQueue<U>(bufferSize);
                    inner.queue = q;
                }
                q.offer(value);
                if (getAndIncrement() != 0) {
                    return;
                }
            }
            //轮询队列，取数据
            drainLoop();
        }

这里需要提一下:MergeObserver 继承了 AtomicInteger，所以这里的tryEmit方法就利用了 AtomicInteger 的同步机制，同时只会有一个 value 被 actual Observer 发射，而且这里刚好可以解答我们上面留下的问题，由于 AtomicInteger CAS锁只能保证操作的原子性，并不保证锁的获取顺序，是抢占式的，所以最终数据的发射顺序并不是固定的(同一个Observable发出的数据是有序的)。
如果没有获取到锁，就会将要发射的数据放入队列中，drainLoop 方法会循环去获取队列中的数据，然后发射。

void drainLoop() {
            //取最终的观察者
            final Observer<? super U> child = this.actual;
            int missed = 1;
            for (;;) {
                //错误检查
                if (checkTerminate()) {
                    return;
                }
                SimplePlainQueue<U> svq = queue;

                if (svq != null) {
                    for (;;) {
                        U o;
                        for (;;) {
                            if (checkTerminate()) {
                                return;
                            }
                            //取数据
                            o = svq.poll();
                            if (o == null) {
                                break;
                            }
                            //交给最后的观察者
                            child.onNext(o);
                        }
                        if (o == null) {
                            break;
                        }
                    }
                }
                .....
                //队列数据轮询完毕,做完成或者错误处理
                if (d && (svq == null || svq.isEmpty()) && n == 0) {
                    Throwable ex = errors.get();
                    if (ex == null) {
                        child.onComplete();
                    } else {
                        child.onError(errors.terminate());
                    }
                    return;
                }
              ....
        }

这个方法就是从循环从数据队列中取数据交给最后的观察者接收。
FlatMap引入CAS机制，在onNext方法中尝试拿到锁，如果拿到则立即交给最终的观察者，否则加入等待队列。
关于map操作法我们先讲到这里，水平有限，难免有纰漏，还望多多指正！

RxJava源码分析(五)变换操作符Map

RxJava源码分析(五)变换操作符Map

引言

map操作符

FlatMap操作符

FlatMap方法

ObservableFlatMap

MergeObserver