如何在 Reactor 中将 Context 与 flatMap() 一起使用？

Question

我无法理解上下文。所以文档说 Context 是：

在组件之间传播的键/值存储，例如 运营商通过上下文协议。上下文是运输的理想选择 正交信息，例如跟踪或安全令牌。

太棒了。

现在让我们假设我们想使用 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

链接您的Publishers，让Context 与您同在

在这种情况下，您连接了所有Publishers（这包括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，这意味着如果父有一个反应器Context - 那么这个上下文将被传播到每个内部@ 987654345@.

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

Answer 2

Reactor-Core v3.4 引入了 Mono.deferContextual 和 Flux.deferContextual，取代了 v3.3 中引入的 Mono.deferWithContext 和 Flux.deferWithContext。

使用这些方法，Oleh Dokukas zip example 可以简化为

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.deferContextual(c -> Mono.just(name)
                .filter(x -> c.hasKey("greetingWord"))
                .map(n -> c.get("greetingWord") + " " + n + " " + "!!!"));
    }
}