在 TimeoutMainSubscriber 的情况下将 Reactor Context 复制到 MDC答案

【问题标题】：Copy Reactor Context to MDC in the case of TimeoutMainSubscriber在 TimeoutMainSubscriber 的情况下将 Reactor Context 复制到 MDC
【发布时间】：2020-02-18 09:08:51
【问题描述】：

我一直在关注https://github.com/archie-swif/webflux-mdc/blob/master/src/main/java/com/example/webfluxmdc/MdcContextLifter.java 的代码，使用钩子将数据从 Reactor Context 放到 MDC。

除了Mono 超时且TimeoutMainSubscriber 正在使用时，该解决方案在大多数情况下都能完美运行。

下面是我使用的小测试。

Mono<Integer> integerMono = Mono.just(1)
            .doOnEach(logger.addToContext(i -> ImmutableMap.of("index", i)))
            .flatMap(___ -> Mono.fromSupplier(() -> {
                try {
                    Thread.sleep(6000);
                } catch (InterruptedException e) {
                    return 0;
                }
                return 1;
            }))
            .timeout(Duration.ofMillis(5000), Schedulers.parallel())
            .doOnEach(logger.info("testMDCLogging"))
            .doOnError(logger.error("testMDCErrorLogging"))
            .subscriberContext(logger.initContext());

当通过 MDC 提升器验证 Context 的传播时，onNext() onError() 和 onComplete() 不会在 TimeoutMainSubscriber 中调用（其中 Context 被复制到 MDC），只有 currentContext() 被调用。

在Mono 超时的情况下，如何传播上下文并复制到MDC 以在doOnError 和doOnEach 中使用？我不想将副本移动到currentContext()，因为这个方法被调用了好几次，在我看来它不是一个最佳的解决方案。

【问题讨论】：

标签： java spring-webflux project-reactor mdc

【解决方案1】：

您是否尝试在 MdcContextLifter 类中覆盖 onError 方法？

    @Override
    public void onError(Throwable t) {
        copyToMdc(coreSubscriber.currentContext()); // similar to onNext()
        coreSubscriber.onError(t);
    }

【讨论】：

不，因为它不起作用。当 Mono 超时时，不会在 TimeoutMainSubscriber 中调用 onNext()、onError() 和 onComplete()。只有 currentContext() 被调用。
@marcterenzi 他们没有被调用，因为上游超时。但是，错误会向下游传播。因此，如果您按照 Nipuna 的建议进行操作，那么它应该会按预期工作。