第三个转换与前一个执行者一起执行

Question

我对在同一方法中使用多个调度器策略有疑问。

Function<Object, Flux<String>> fetchGreekGods = obj -> {

    return Flux.just(this.config.getApiMap().get(GREEK))
            .publishOn(Schedulers.immediate())
            .map(toURL)
            .map(fetch)
            .flatMap(serializeFlux)
            .log();
};

Function<Flux<String>, Flux<Tuple2<String, Integer>>> fetchWikipediaGodInfo = god -> {

    return Flux.from(god)
            .publishOn(Schedulers.elastic())
            .map(str -> {
                return new Tuple2<String, Integer>(str, generateWikiAddress
                        .andThen(toURL)
                        .andThen(fetch)
                        .andThen(String::length)
                        .apply(str));
            })
            .log();
};

Function<Flux<Tuple2<String, Integer>>, Flux<String>> max = godInfo -> {

    return Flux.from(godInfo)
            .publishOn(Schedulers.immediate())
            .sort(Comparator.comparing(Tuple2::_2))
            .takeLast(1)
            .map(t -> t._1)
            .log();
};

public Mono<String> reactorSolution() {

        return Flux.empty()
            .transform(fetchGreekGods)
            .transform(fetchWikipediaGodInfo)
            .transform(max)
            .next();
}

为什么如果我为每个转换定义不同的调度程序，在最后一个，在 Max 的执行中，在日志中，我观察到上一个的策略：fetchWikipediaGodInfo

在逻辑上，我想要max，代码使用主线程，但目前，我观察前一个。

2019-09-22 18:08:30 [elastic-2] INFO  reactor.Flux.MapFuseable.3 - | onNext(Apollo)
2019-09-22 18:08:30 [elastic-2] INFO  reactor.Flux.MapFuseable.3 - | cancel()
2019-09-22 18:08:30 [elastic-2] INFO  reactor.Flux.MapFuseable.3 - | onComplete()

理论上，第三次变换应该在主线程中运行，但目前该方法在弹性中运行但不正确，因为我明确定义在主线程中运行。

如何解决？

在此先感谢

胡安·安东尼奥

Answer 1

根据有关Schedulers的文档

Scheduler 类具有静态方法，可以访问 以下执行上下文：

• 当前线程 (Schedulers.immediate())。

当前线程不一定是主线程，它很可能是最后一个线程。并且之前的变换是使用弹性的。