WebFlux：从流内对象添加订阅服务器上下文值

Question

我有一个Webflux应用程序，我正在使用订阅服务器上下文填充MDC值，以便它们向下游传播。我已经实现了这个项目中的类来处理订阅者之间的MDC传输，设置webfilter来向传入的请求添加请求ID，并且可以在日志中看到请求ID作为MDC的一部分。

public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
    return chain.filter(exchange)
            .contextWrite(Context.of("requestId", UUID.randomUUID().toString()));
}

日志输出：

2021-09-30 17:15:29,963 [reactor-tcp-nio-2] INFO  c.p.l.s.RepoService:33 - MDC[requestId=ec0b68cf-ba4d-4c7f-afa4-f67fc97ebcbf] - Found user

public Mono<Response> doProcessing(String userId, Object object) {
    return userRepo.findUserById(userId) //Returns UserEntity object
            .flatMap(userEntity -> service.doMoreProcessing(userEntity, object))
            .contextWrite(Context.of("userId", userId, "email", userEntity.getEmail()));
}

public Mono<Response> doProcessing(String userId, Object object) {
    return userRepo.findUserById(userId) //Returns UserEntity object
            .flatMap(userEntity -> Mono.deferContextual(ctx -> Mono.just(userEntity).contextWrite(Context.of(ctx).putAll(Context.of("userId", userId, "email", userEntity.getEmail()).readOnly()))))
            .flatMap(userEntity -> service.doMoreProcessing(userEntity, object));
}

public Mono<Response> doProcessing(String userId, Object object) {
    return userRepo.findUserById(userId) //Returns UserEntity object
            .transformDeferredContextual((userEntityMono, contextView) -> userEntityMono.flatMap(userEntity -> Mono.just(userEntity).contextWrite(Context.of(contextView).putAll(Context.of("userId", userId, "email", userEntity.getEmail()).readOnly()))))
            .flatMap(userEntity -> service.doMoreProcessing(userEntity, object));
}

...但是什么都没起作用。如何从属于进程流的对象向订阅服务器上下文添加数据？

Answer 1

你可以试试这样的东西

public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        return chain.filter(exchange)
                .contextWrite(Context.of("your-key", new HashMap<String, Object>()));
}

public Mono<Response> doProcessing(String userId, Object object) {
    return userRepo.findUserById(userId)
            .doOnEach(signal -> {
                if (!signal.isOnNext()) return;
                Optional<Map<String, Object>> optional = signal.getContextView().getOrEmpty("your-key");
                optional.ifPresent(map -> {
                    map.put("userId", signal.get().getId());
                    map.put("email", signal.get().getEmail());
                });
            });
}