java - 如何在 Reactor 中使用 Context 和 flatMap()？

Question

我在理解上下文时遇到问题。所以文档说 Context 是:

A key/value store that is propagated between components such as operators via the context protocol. Contexts are ideal to transport orthogonal information such as tracing or security tokens.

很好。

现在让我们假设我们想要使用 Context 传播一些东西，让它无处不在。要调用另一个异步代码，我们只需使用 flatMap() 方法。

问题:如何在调用的方法中访问上下文？

示例(简单)代码:

public class TestFlatMap {
    public static void main(final String ...args) {
        final Flux<String> greetings = Flux.just("Hubert", "Sharon")
            .flatMap(TestFlatMap::nameToGreeting)
            .subscriberContext(context ->
                Context.of("greetingWord", "Hello")  // context initialized
            );
        greetings.subscribe(System.out::println);
    }

    private static Mono<String> nameToGreeting(final String name) {
        return Mono.just("Hello " + name + " !!!");  // ALERT: we don't have Context here
    }
}

被调用的方法可以(而且很可能会)在另一个类中。

提前感谢您的帮助!

编辑:删除了一些代码，使问题更加简洁明了。

Answer 1

链接你的 Publisher 并且 Context 可能与你同在

在这种情况下，您连接了所有的 Publisher(这包括 flatMap/concatMap 和类似运算符中的连接)，您将在整个流运行时中正确传播 Context。

要在 nameToGreeting 方法中访问 Context，您可以调用 Mono.subscribeContext 并检索存储的信息事件，如果方法似乎不是有关的。下面显示了提到的概念:

public class TestFlatMap {
    public static void main(final String ...args) {
        final Flux<String> greetings = Flux.just("Hubert", "Sharon")
                                           .flatMap(TestFlatMap::nameToGreeting)
                                           .subscriberContext(context ->
                                                   Context.of("greetingWord", "Hello")  // context initialized
                                           );
        greetings.subscribe(System.out::println);
    }

    private static Mono<String> nameToGreeting(final String name) {
        return Mono.subscriberContext()
                   .filter(c -> c.hasKey("greetingWord"))
                   .map(c -> c.get("greetingWord"))
                   .flatMap(greetingWord -> Mono.just(greetingWord + " " + name + " " + "!!!"));// ALERT: we have Context here !!!
    }
}

此外，您还可以通过以下方式执行相同操作，使用 zip 运算符，稍后合并结果:

public class TestFlatMap {
    public static void main(final String ...args) {
        final Flux<String> greetings = Flux.just("Hubert", "Sharon")
                                           .flatMap(TestFlatMap::nameToGreeting)
                                           .subscriberContext(context ->
                                                   Context.of("greetingWord", "Hello")  // context initialized
                                           );
        greetings.subscribe(System.out::println);
    }

    private static Mono<String> nameToGreeting(final String name) {
        return Mono.zip(
            Mono.subscriberContext()
                .filter(c -> c.hasKey("greetingWord"))
                .map(c -> c.get("greetingWord")), // ALERT: we have Context here !!!
            Mono.just(name),
            (greetingWord, receivedName) -> greetingWord + " " + receivedName + " " + "!!!"
        );
    }
}

那么，它为什么有效？

正如我们从上面的示例中看到的，nameToGreeting 是在主 Flux 的上下文中调用的。在引擎盖下 -> (Here Some FluxFlatMap internals) ，每个映射的 Publisher 都被 FlatMapInner 订阅。如果我们看一下 FlatMapInner并寻找 the currentContext override我们将看到，FlatMapInner 使用父级 Context，这意味着如果父级有一个 Reactor Context - 那么这个上下文将传播到每个内部发布者。

因此，nameToGreeting 方法返回的Mono 将具有与其父级相同的Context