Flux.just("tom", "jack", "allen")
                .map(s-> s.concat("@qq.com"))
                .filter(s -> s.length() >= 11)
                .subscribe(System.out::println);

结果：

jack@qq.com
allen@qq.com

1.声明阶段

Flux#just()

Flux#just()

• Flux#fromArray(T[] array)
• onAssembly(new FluxArray<>(array))
• 最后返回的就是FluxArray<>(array)

final class FluxArray<T> extends Flux<T> implements Fuseable, SourceProducer<T> {

    final T[] array;

    @SafeVarargs
    public FluxArray(T... array) {
        this.array = Objects.requireNonNull(array, "array");
    }

Flux#map(Function mapper)

Flux#map(Function<? super T, ? extends V> mapper)

• onAssembly(new FluxMapFuseable<>(this, mapper))

final class FluxMapFuseable<T, R> extends InternalFluxOperator<T, R> implements Fuseable {

    final Function<? super T, ? extends R> mapper;

    FluxMapFuseable(Flux<? extends T> source,
            Function<? super T, ? extends R> mapper) {
        super(source);
        this.mapper = Objects.requireNonNull(mapper, "mapper");
    }

Flux#filter

Flux#filter(Predicate<? super T> p)

• onAssembly(new FluxFilterFuseable<>(this, p))

2.subscribe

2.1 subscribe阶段

Flux#subscribe(Consumer)

Flux#subscribe(Consumer)

• subscribe(consumer, null, null)
• subscribe(consumer, errorConsumer, completeConsumer, (Context) null)
• subscribeWith(new LambdaSubscriber<>(consumer, errorConsumer, completeConsumer, null, initialContext))
• subscribe(Subscriber<? super T> actual)

    public final void subscribe(Subscriber<? super T> actual) {
        CorePublisher publisher = Operators.onLastAssembly(this);
        CoreSubscriber subscriber = Operators.toCoreSubscriber(actual);

        if (subscriber instanceof Fuseable.QueueSubscription && this != publisher && this instanceof Fuseable && !(publisher instanceof Fuseable)) {
            subscriber = new FluxHide.SuppressFuseableSubscriber<>(subscriber);
        }

        try {
            if (publisher instanceof OptimizableOperator) {
                OptimizableOperator operator = (OptimizableOperator) publisher;
                while (true) {
                    subscriber = operator.subscribeOrReturn(subscriber);
                    if (subscriber == null) {
                        // null means "I will subscribe myself", returning...
                        return;
                    }
                    OptimizableOperator newSource = operator.nextOptimizableSource();
                    if (newSource == null) {
                        publisher = operator.source();
                        break;
                    }
                    operator = newSource;
                }
            }

            publisher.subscribe(subscriber);
        }
        catch (Throwable e) {
            Operators.reportThrowInSubscribe(subscriber, e);
            return;
        }
    }

Flux#subscribe(Subscriber actual)

• 这里this是FluxFilterFuseable，入参actual是LambdaSubscriber
• subscriber = operator.subscribeOrReturn(subscriber)，这里operator是FluxFilterFuseable，进入FluxFilterFuseable#subscribeOrReturn，创建FilterFuseableSubscriber<>(actual, predicate)
• newSource = operator.nextOptimizableSource()获取FluxFilterFuseable的OptimizableOperator，其实就是FluxMapFuseable，然后赋值给operator
• 循环调用subscriber = operator.subscribeOrReturn(subscriber)，进入FluxMapFuseable#subscribeOrReturn，创建MapFuseableConditionalSubscriber<>(cs, mapper)
• newSource = operator.nextOptimizableSource()获取FluxMapFuseable的OptimizableOperator，为null。若为null，则获取publisher = operator.source()也即FluxArray，然后break，此时已经跳出循环
• publisher.subscribe(subscriber)，此时publisher是最里层的FluxArray，subscriber是MapFuseableConditionalSubscriber
• FluxArray#subscribe(actual)
• subscribe(actual, array)

FluxArray#subscribe(CoreSubscriber<? super T> s, T[] array)

    public static <T> void subscribe(CoreSubscriber<? super T> s, T[] array) {
        if (array.length == 0) {
            Operators.complete(s);
            return;
        }
        if (s instanceof ConditionalSubscriber) {
            s.onSubscribe(new ArrayConditionalSubscription<>((ConditionalSubscriber<? super T>) s, array));
        }
        else {
            s.onSubscribe(new ArraySubscription<>(s, array));
        }
    }

2.2 onSubscribe阶段

FluxArray#subscribe(CoreSubscriber<? super T> s, T[] array)

• s是MapFuseableConditionalSubscriber，array是数据源
• s.onSubscribe(new ArrayConditionalSubscription<>((ConditionalSubscriber<? super T>) s, array))
• MapFuseableConditionalSubscriber#onSubscribe(Subscription s)，将this.s = (QueueSubscription<T>) s，并且调用actual.onSubscribe(this)，actual是FilterFuseableSubscriber
• FilterFuseableSubscriber#onSubscribe(Subscription s)，这里s是MapFuseableConditionalSubscriber，将this.s赋值为MapFuseableConditionalSubscriber，并且调用actual.onSubscribe(this)，actual是LambdaSubscriber
• LambdaSubscriber#onSubscribe(Subscription s)，这里s是FilterFuseableSubscriber，赋值this.subscription为FilterFuseableSubscriber，然后调用s.request(Long.MAX_VALUE)

LambdaSubscriber#onSubscribe

    public final void onSubscribe(Subscription s) {
        if (Operators.validate(subscription, s)) {
            this.subscription = s;
            if (subscriptionConsumer != null) {
                try {
                    subscriptionConsumer.accept(s);
                }
                catch (Throwable t) {
                    Exceptions.throwIfFatal(t);
                    s.cancel();
                    onError(t);
                }
            }
            else {
                s.request(Long.MAX_VALUE);
            }
        }
    }

2.3 request阶段

FilterFuseableSubscriber#request(Long.MAX_VALUE)

• s.request(n)，也即调用MapFuseableConditionalSubscriber#request
• s.request(n)，也即调用ArrayConditionalSubscription#request

ArrayConditionalSubscription#request

        public void request(long n) {
            if (Operators.validate(n)) {
                if (Operators.addCap(REQUESTED, this, n) == 0) {
                    if (n == Long.MAX_VALUE) {
                        fastPath();
                    }
                    else {
                        slowPath(n);
                    }
                }
            }
        }

2.4 调用阶段

ArrayConditionalSubscription#request

• ArrayConditionalSubscription#fastPath()
• 这里循环处理array中每一个元素，调用actual.onNext(t)，这里actual是MapFuseableConditionalSubscriber，也即MapFuseableConditionalSubscriber#onNext()
• 调用v = mapper.apply(t)，真的mapper逻辑，然后调用actual.onNext(v)，是FilterFuseableSubscriber#onNext(v)
• 调用b = predicate.test(t)，真正的Filter逻辑，if (b) {actual.onNext(t)}，否则s.request(1)，丢弃该不满足条件的元素。这里actual是LambdaSubscriber，也即LambdaSubscriber#onNext
• 最终调用consumer.accept(x)，也就是最终的消费者进行处理

执行逻辑的核心在ArrayConditionalSubscription#fastPath()

        void fastPath() {
            final T[] a = array;
            final int len = a.length;
            final Subscriber<? super T> s = actual;

            for (int i = index; i != len; i++) {
                if (cancelled) {
                    return;
                }

                T t = a[i];

                if (t == null) {
                    s.onError(new NullPointerException("The " + i + "th array element was null"));
                    return;
                }

                s.onNext(t);
            }
            if (cancelled) {
                return;
            }
            s.onComplete();
        }

总结

• 1）声明阶段

  
  

• 2）subscribe阶段

  
  

• 3）onSubscribe阶段
  在此处将 Publisher 和 Subscriber 包裹成一个 Subscription 对象，为每一层Subscriber添加Subscription

  
  

• 4）request阶段
  最终在 原始 Subscriber 对象（最外层，也就是FluxArray中的ArrayConditionalSubscription）调用 request() ，触发 Subscription 的 Source 获得数据作为 onNext 的参数。

• 5）调用阶段
  遍历处理数组的每个元素，从最外层的Subscription开始往里层调用onNext()，每一层需要调用对应的实际处理逻辑，比如mapper/ predicate/ consume等。

参考

• 1）http://blog.yanick.site/2018/06/26/java/spring/projectreactor/