在 java 反应堆中使用 objectmapper in flux 不适当的阻塞方法调用时收到警告

Question

我是 reactor 的新手，我试图从 Iterable 创建一个 flux。然后我想通过使用对象映射器将我的对象转换为字符串。然后 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);
                });

Answer 1

你需要这样做：

     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);
     });

Answer 2

我会给你一个答案，但我不太确定这是你想要的。似乎阻塞了线程。所以如果你阻塞线程，你就无法获得反应式的确切好处。这就是 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 对象并不重要。（如果你不改变）。你的情况没有必要。

在 java 反应堆中使用 objectmapper in flux 不适当的阻塞方法调用时收到警告

Getting a warning when use objectmapper in flux inappropriate blocking method call in java reactor

reactor

objectmapper

spring-webflux