spring-boot - 在 doOnSubscribe、doOnSuccess 和 doOnError 中获取 Reactor 订阅者上下文

Question

我正在尝试实现 Reactor Subscriber Context ( http://projectreactor.io/docs/core/release/reference/#context )，这样我就可以将 SLF4J MDC 中的值传递到 Flux 中，然后我可以使用这些值进行日志记录。

我使用subscriberContext() 方法来设置如下值:

someFlux().subscriberContext(Context.of(MDC_ATTRIBUTE, MDC.get(MDC_ATTRIBUTE)));

我还可以访问链中的上下文。例如使用 flatMap:

.flatMap(r -> Mono.subscriberContext().map(ctx -> {
    String name = ctx.getOrDefault(MDC_ATTRIBUTE_NAME, "NO CTX");
    return r;
}))

还有 doOnEach()作品:

.doOnEach(signal -> {
    Context ctx = signal.getContext();
    if (signal.isOnNext()) {
        try (MDC.MDCCloseable closeable = MDC.putCloseable(MDC_ATTRIBUTE_NAME, ctx.getOrDefault(MDC_ATTRIBUTE_NAME, "MAAAAN"))) {
            log.debug("FINISHED: {}", requestName);
        }
    }
})

它只有一个问题。我想登录 doOnSubscribe , doOnError并在 doOnSuccess .虽然我可以使用 doOnEach 检查 signal.isOnNext()或 signal.isOnComplete() ，我发现 signal.isOnSubscribe()永远不会被调用。

所以 问题是 : 我如何在 doOnSubscribe() 中获取上下文或者这根本不可能？

Answer 1

这是可能的，不是在 100% 的用例中，并且有一些技巧:

Flux.just("foo")
    .doOnSubscribe(sub -> {
        Scannable actual = Scannable.from(sub).scan(Scannable.Attr.ACTUAL);
        if (actual instanceof CoreSubscriber) {
            Context context = ((CoreSubscriber) actual).currentContext();

            System.out.println(context);
        }
    })
    .map(v -> "value: " + v) //below or above doOnSubscribe is fine
    .subscriberContext(Context.of("foo", "bar")) //MUST be below doOnSubscribe
    .blockLast();