Spring boot + webflux：并行运行某些步骤时上下文丢失

Posted 2023-02-27

【中文标题】Spring boot + webflux：并行运行某些步骤时上下文丢失

【英文标题】：Spring boot + webflux: context lost when running some steps in parallel

【发布时间】：2019-08-31 13:40:43

【问题描述】：

春季启动：2.1.3.RELEASE

你好，

我正在尝试使用 spring webflux 的上下文特性来携带一个简单的变量。我有一个 WebFilter 设置具有此类变量的上下文，并且我尝试在我的通量/流的不同阶段在我的控制器中使用它。在某些时候，我在调用 Flux 类的方法“parallel()”后丢失了它。

过滤器：

public class TestFilter implements WebFilter 

    private Logger LOG = LoggerFactory.getLogger(TestFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) 
        return chain.filter(exchange)
            .doOnEach(voidSignal -> System.out.println("filter:"+voidSignal.getContext().getOrEmpty("blob"))).subscriberContext(Context.of("blob", "kapoue"));
    

控制器：

@RestController
@RequestMapping(TestControllerWebFlux.ROOT)
public class TestControllerWebFlux 

    static final String ROOT = "/flux";
    static final String TEST = "/test";

    private WebClient webClient = WebClient.create();

    @GetMapping(
            value = TEST,
            produces = MediaType.APPLICATION_JSON_VALUE)
    public Mono<String> test() 
        System.out.println("controller1:"+Thread.currentThread());

        Flux<String> call = webClient.get().uri("http://localhost:" + 8080 + ROOT + "/test2").retrieve().bodyToFlux(Result.class).map(Result::getValue);

        return call.map(s -> s+"0")
            .doOnEach(stringSignal -> System.out.println("controller2:"+stringSignal.getContext().getOrEmpty("blob")))
            .parallel()
            .doOnEach(stringSignal -> System.out.println("controller3:"+stringSignal.getContext().getOrEmpty("blob")))
            .map(s -> s+"0")
            .doOnEach(stringSignal -> System.out.println("controller4:"+stringSignal.getContext().getOrEmpty("blob")))
            .reduce((s, s2) -> s+s2)
            .doOnEach(stringSignal -> System.out.println("controller5:"+stringSignal.getContext().getOrEmpty("blob")))
            .map(s -> 
                System.out.println("controller6:"+Thread.currentThread());
                return s;
            );
    

    @GetMapping(
        value = "test2",
        produces = MediaType.APPLICATION_JSON_VALUE)
    public Flux<Result> test2() 
        return Flux.just(new Result("0"), new Result("0"), new Result("0"));
    

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Result 
        private String value;
    

我所做的只是调用http://localhost:8080/flux/test/ 端点，我明白了：

controller1:Thread[reactor-http-nio-2,5,main] 控制器2：可选[kapoue] 控制器3：可选的.empty 控制器4：可选的.empty 控制器2：可选[kapoue] 控制器3：可选的.empty 控制器4：可选的.empty 控制器2：可选[kapoue] 控制器3：可选的.empty 控制器4：可选的.empty 控制器2：可选[kapoue] 控制器3：可选的.empty 控制器4：可选的.empty 控制器3：可选的.empty 控制器4：可选的.empty 控制器3：可选的.empty 控制器4：可选的.empty 控制器3：可选的.empty 控制器4：可选的.empty 控制器5：可选[kapoue] controller6:Thread[reactor-http-nio-2,5,main] 过滤器：可选[kapoue]

如您所见，上下文在“并行”方法之后立即丢失，并在减少后以某种方式返回。

这是一个错误还是我不应该在这样的调用之后尝试并行运行？

提前感谢您的帮助。

【参考方案1】：

这看起来像是 Reactor 中的错误。我举报了：https://github.com/reactor/reactor-core/issues/1656

【讨论】：

... 并在github.com/reactor/reactor-core/pull/1657 中修复。上下文没有丢失，只是 .doOnEach 没有设置在您收到的 Signal 对象上。