RxJava源码解析04-变换过程(flatmap)

太过爱你忘了你带给我的痛 2023-10-17 15:46 59阅读 0赞

#### flatmap ####

public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
        //...
        return merge(map(func));
    }

在flatmap中，首先会调用`map()`将func转换为`Observable`类型，由于这里传入的泛型是`Observable<? extends R>>`，故会返回`Observable<Observable<? extends R>>>`类型，具体的转换在上一篇文章上已经讲过了。

获得`Observable<Observable<? extends R>>>`后，会调用`merge()`，来看下merge操作符的定义

将多个Observables的输出合并，就好像它们是一个单个的Observable一样。

所以这里也就是把原来的Observable和`map()`新生成的Observable进行合并，从而生成新的Observable。

直接来看merge的核心代码

//Observable类
    public static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
        //...
        return source.lift(OperatorMerge.<T>instance(false));
    }
    
    public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return unsafeCreate(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

可以看到`lift()`传参的时候，调用了`OperatorMerge.<T>instance(false)`，这个最终会返回OperatorMerge实例。  
这个类我们之后再来看。

再回到`Observable#lift()`

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return unsafeCreate(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

这里会把传入onSubscribe和operator生成一个OnSubscribeLift 。OnSubscribeLift 会把数据先交由`operator#call()`调用后，再给onSubscribe调用，这个我们上一篇文章中已经知道了。

最终通过unsafeCreate返回一个新的Observable。

主要逻辑在`operator#call()`

public final class OperatorMerge<T> implements Operator<T, Observable<? extends T>> {
        @Override
        public Subscriber<Observable<? extends T>> call(final Subscriber<? super T> child) {
            MergeSubscriber<T> subscriber = new MergeSubscriber<T>(child, delayErrors, maxConcurrent);
            MergeProducer<T> producer = new MergeProducer<T>(subscriber);
            subscriber.producer = producer;
    
            child.add(subscriber);
            child.setProducer(producer);
    
            return subscriber;
        }
    
        //...
    }

call方法这里会创建一个MergeSubscriber，并且会调用`setProducer()`

public void setProducer(Producer p) {
        //...
        if (toRequest == NOT_SET) {
            producer.request(Long.MAX_VALUE);
        } else {
            producer.request(toRequest);
        }
    }

即调用MergeProducer的request()

public void request(long n) {
            if (n > 0) {
                if (get() == Long.MAX_VALUE) {
                    return;
                }
                BackpressureUtils.getAndAddRequest(this, n);
                subscriber.emit();
            } else if (n < 0) {
                throw new IllegalArgumentException("n >= 0 required");
            }
        }

可以看到，最终调用了MergeSubscriber的emit()。emit会从queue中不断获取数据t并传给Subscriber child去执行。从而把Observable的数据逐个交给Subscriber的OnNext、onCompleted、onError进行调用。

void emitLoop() {
            boolean skipFinal = false;
            try {
                final Subscriber<? super T> child = this.child;
                for (;;) {
                    // eagerly check if child unsubscribed or we reached a terminal state.
                    if (checkTerminate()) {
                        skipFinal = true;
                        return;
                    }
                    Queue<Object> svq = queue;
    
                    long r = producer.get();
                    boolean unbounded = r == Long.MAX_VALUE;
    
                    // count the number of 'completed' sources to replenish them in batches
                    int replenishMain = 0;
    
                    // try emitting as many scalars as possible
                    if (svq != null) {
                        for (;;) {
                            int scalarEmission = 0;
                            Object o = null;
                            while (r > 0) {
                                o = svq.poll();
                                // eagerly check if child unsubscribed or we reached a terminal state.
                                if (checkTerminate()) {
                                    skipFinal = true;
                                    return;
                                }
                                if (o == null) {
                                    break;
                                }
                                T v = NotificationLite.getValue(o);
                                // if child throws, report bounce it back immediately
                                try {
                                    child.onNext(v);
                                } catch (Throwable t) {
                                    if (!delayErrors) {
                                        Exceptions.throwIfFatal(t);
                                        skipFinal = true;
                                        unsubscribe();
                                        child.onError(t);
                                        return;
                                    }
                                    getOrCreateErrorQueue().offer(t);
                                }
                                replenishMain++;
                                scalarEmission++;
                                r--;
                            }
                            if (scalarEmission > 0) {
                                if (unbounded) {
                                    r = Long.MAX_VALUE;
                                } else {
                                    r = producer.produced(scalarEmission);
                                }
                            }
                            if (r == 0L || o == null) {
                                break;
                            }
                        }
                    }
    
                    /*
                     * We need to read done before innerSubscribers because innerSubscribers are added
                     * before done is set to true. If it were the other way around, we could read an empty
                     * innerSubscribers, get paused and then read a done flag but an async producer
                     * might have added more subscribers between the two.
                     */
                    boolean d = done;
                    // re-read svq because it could have been created
                    // asynchronously just before done was set to true.
                    svq = queue;
                    // read the current set of inner subscribers
                    InnerSubscriber<?>[] inner = innerSubscribers;
                    int n = inner.length;
    
                    // check if upstream is done, there are no scalar values
                    // and no active inner subscriptions
                    if (d && (svq == null || svq.isEmpty()) && n == 0) {
                        Queue<Throwable> e = errors;
                        if (e == null || e.isEmpty()) {
                            child.onCompleted();
                        } else {
                            reportError();
                        }
                        skipFinal = true;
                        return;
                    }
    
                    boolean innerCompleted = false;
                    if (n > 0) {
                        // let's continue the round-robin emission from last location
                        long startId = lastId;
                        int index = lastIndex;
    
                        // in case there were changes in the array or the index
                        // no longer points to the inner with the cached id
                        if (n <= index || inner[index].id != startId) {
                            if (n <= index) {
                                index = 0;
                            }
                            // try locating the inner with the cached index
                            int j = index;
                            for (int i = 0; i < n; i++) {
                                if (inner[j].id == startId) {
                                    break;
                                }
                                // wrap around in round-robin fashion
                                j++;
                                if (j == n) {
                                    j = 0;
                                }
                            }
                            // if we found it again, j will point to it
                            // otherwise, we continue with the replacement at j
                            index = j;
                            lastIndex = j;
                            lastId = inner[j].id;
                        }
    
                        int j = index;
                        // loop through all sources once to avoid delaying any new sources too much
                        for (int i = 0; i < n; i++) {
                            // eagerly check if child unsubscribed or we reached a terminal state.
                            if (checkTerminate()) {
                                skipFinal = true;
                                return;
                            }
                            @SuppressWarnings("unchecked")
                            InnerSubscriber<T> is = (InnerSubscriber<T>)inner[j];
    
                            Object o = null;
                            for (;;) {
                                int produced = 0;
                                while (r > 0) {
                                    // eagerly check if child unsubscribed or we reached a terminal state.
                                    if (checkTerminate()) {
                                        skipFinal = true;
                                        return;
                                    }
                                    RxRingBuffer q = is.queue;
                                    if (q == null) {
                                        break;
                                    }
                                    o = q.poll();
                                    if (o == null) {
                                        break;
                                    }
                                    T v = NotificationLite.getValue(o);
                                    // if child throws, report bounce it back immediately
                                    try {
                                        child.onNext(v);
                                    } catch (Throwable t) {
                                        skipFinal = true;
                                        Exceptions.throwIfFatal(t);
                                        try {
                                            child.onError(t);
                                        } finally {
                                            unsubscribe();
                                        }
                                        return;
                                    }
                                    r--;
                                    produced++;
                                }
                                if (produced > 0) {
                                    if (!unbounded) {
                                        r = producer.produced(produced);
                                    } else {
                                        r = Long.MAX_VALUE;
                                    }
                                    is.requestMore(produced);
                                }
                                // if we run out of requests or queued values, break
                                if (r == 0 || o == null) {
                                    break;
                                }
                            }
                            boolean innerDone = is.done;
                            RxRingBuffer innerQueue = is.queue;
                            if (innerDone && (innerQueue == null || innerQueue.isEmpty())) {
                                removeInner(is);
                                if (checkTerminate()) {
                                    skipFinal = true;
                                    return;
                                }
                                replenishMain++;
                                innerCompleted = true;
                            }
                            // if we run out of requests, don't try the other sources
                            if (r == 0) {
                                break;
                            }
    
                            // wrap around in round-robin fashion
                            j++;
                            if (j == n) {
                                j = 0;
                            }
                        }
                        // if we run out of requests or just completed a round, save the index and id
                        lastIndex = j;
                        lastId = inner[j].id;
                    }
    
                    if (replenishMain > 0) {
                        request(replenishMain);
                    }
                    // if one or more inner completed, loop again to see if we can terminate the whole stream
                    if (innerCompleted) {
                        continue;
                    }
                    // in case there were updates to the state, we loop again
                    synchronized (this) {
                        if (!missed) {
                            skipFinal = true;
                            emitting = false;
                            break;
                        }
                        missed = false;
                    }
                }
            } finally {
                if (!skipFinal) {
                    synchronized (this) {
                        emitting = false;
                    }
                }
            }
        }

至此，完成了flatmap变换过程的解析。

#### 总结 ####

flatmap的变换实际上就是先通过`map()`将Fun1转换为`Observable< Observable<T>>`，再交由`merge()`将两个Observable合并，生成新的Observable。当执行时，内部会通过不断获取队列的数据，再交由Subscriber进行OnNext、onCompleted、onError的调用。