在 Java 反应器中的通量不适当的阻塞方法调用中使用 objectmapper 时收到警告答案

【问题标题】：Getting a warning when use objectmapper in flux inappropriate blocking method call in java reactor在 Java 反应器中的通量不适当的阻塞方法调用中使用 objectmapper 时收到警告
【发布时间】：2021-09-20 01:23:41
【问题描述】：

我是 reactor 的新手，我试图从 Iterable 创建一个通量。然后我想使用对象映射器将我的对象转换为字符串。然后 ide 在这部分代码new ObjectMapper().writeValueAsString(event) 中警告这样的消息。消息Inappropriate blocking method call。没有编译错误。你能提出一个解决方案吗？


        Flux.fromIterable(Arrays.asList(new Event(), new Event()))
                .flatMap(event -> {
                    try {
                        return Mono.just(new ObjectMapper().writeValueAsString(event));
                    } catch (JsonProcessingException e) {
                        return Mono.error(e);
                    }
                })
                .subscribe(jsonStrin -> {
                    System.out.println("jsonStrin = " + jsonStrin);
                });

【问题讨论】：

标签： spring-webflux objectmapper reactor

【解决方案1】：

我会给你一个答案，但我不太确定这是你想要的。似乎阻塞了线程。因此，如果您阻塞线程，您将无法获得反应式的确切好处。这就是 IDE 警告您的原因。您可以使用monoSink 创建单声道。如下所示。

        AtomicReference<ObjectMapper> objectMapper = new AtomicReference<>(new ObjectMapper());
        Flux.fromIterable(Arrays.asList(new Event(), new Event()))
                .flatMap(event -> {
                    return Mono.create(monoSink -> {
                        try {
                            monoSink.success(objectMapper .writeValueAsString(event));
                        } catch (JsonProcessingException e) {
                            monoSink.error(e);
                        }
                    });

                })
                .cast(String.class) // this cast will help you to axact data type that you want to continue the pipeline
                .subscribe(jsonString -> {
                    System.out.println("jsonString = " + jsonString);
                });

请尝试此方法并检查错误是否消失。

objectMapper 是否像您一样是普通的 java 对象并不重要。（如果你不改变）。对于您的情况，这不是必需的。

【讨论】：

这很难看，不是解决方案。阻塞调用应该放在它们自己的调度器上，这样它们就不会阻塞执行。应避免原子引用。投反对票。
@Toerktumlare 那么，您对 OP 的逻辑有何建议？
阻塞调用应该按照参考文档projectreactor.io/docs/core/release/reference/…处理。如果你使用AtomicReferences，你在线程之间共享内存，因此创建状态，线程和保持状态是非常糟糕的。 OP 提供的信息太少，所以应该询问有关服务、消费者等的更多信息。这是 Spring Boot 服务吗？ api是什么样的？等等，它需要更多的上下文。一旦你需要在反应器中使用原子引用，它就会产生直接的代码味道。

【解决方案2】：

你需要这样做：

     Flux.fromIterable(Arrays.asList(new Event(), new Event()))
     .flatMap(event -> {
        try {
             return Mono.just(new ObjectMapper().writeValueAsString(event));
        } catch (JsonProcessingException e) {
            return Mono.error(e);
        }
     })
     .subscribe(jsonStrin -> {
         System.out.println("jsonStrin = " + jsonStrin);
     });

【讨论】：

我的代码和你登顶的代码有什么不同？我看不到反抗？
我发布的代码对我来说很好。