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

Question

我一直在按照 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 提升器验证上下文传播时，onNext() onError() 和 onComplete() 未在 TimeoutMainSubscriber 中调用（上下文被复制到 MDC） , 只调用 currentContext()。

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

Answer 1

您是否尝试覆盖 MdcContextLifter 中的 onError 方法 class？

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

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

Copy Reactor Context to MDC in the case of TimeoutMainSubscriber

java

mdc

project-reactor

spring-webflux