WebFlux：为什么我需要在 CRUD 中使用 flatMap

Question

我在 Internet 上找到了示例，但这并没有让我完全理解。 使用 WebFlux 时的标准 CRUD。

路由器：

@Configuration
public class PersonRouter {

    @Bean
    public RouterFunction<ServerResponse> route(PersonHandler handler) {
        return RouterFunctions
                .route(GET("/getAllPersons").and(accept(MediaType.APPLICATION_JSON)), handler::findAll)
                .andRoute(GET("/getPerson/{id}").and(accept(MediaType.APPLICATION_STREAM_JSON)), handler::findById)
                .andRoute(POST("/createPerson").and(accept(MediaType.APPLICATION_JSON)), handler::save)
                .andRoute(DELETE("/deletePerson/{id}").and(accept(MediaType.APPLICATION_JSON)), handler::delete);
    }

}

处理程序：

@Component
public class PersonHandler {

    private final PersonService personService;

    public PersonHandler(PersonService personService) {
        this.personService = personService;
    }

    public Mono<ServerResponse> findById(ServerRequest request) {
        String id = request.pathVariable("id");
        return ok()
                .contentType(MediaType.APPLICATION_JSON)
                .body(personService.getById(id), Person.class);
    }

    public Mono<ServerResponse> findAll(ServerRequest request) {
        return ok()
                .contentType(MediaType.APPLICATION_JSON)
                .body(personService.getAll(), Person.class);
    }

    public Mono<ServerResponse> save(ServerRequest request) {
        final Mono<Person> person = request.bodyToMono(Person.class);
        return ok()
                .contentType(MediaType.APPLICATION_JSON)
                .body(fromPublisher(person.flatMap(personService::save), Person.class));
    }

    public Mono<ServerResponse> delete(ServerRequest request) {
        String id = request.pathVariable("id");
        return ok()
                .contentType(MediaType.APPLICATION_JSON)
                .body(personService.delete(id), Void.class);
    }

}

存储库：

@Repository
public interface PersonRepository extends ReactiveMongoRepository<Person, String> {
}

服务：

@Service
@Transactional
@AllArgsConstructor
public class PersonService {

    private final PersonRepository personRepository;

    public Flux<Person> getAll() {
        return personRepository.findAll().switchIfEmpty(Flux.empty());
    }

    public Mono<Person> getById(final String id) {
        return personRepository.findById(id);
    }

    public Mono update(final String id, final Person person) {
        return personRepository.save(person);
    }

    public Mono save(final Person person) {
        return personRepository.save(person);
    }

    public Mono delete(final String id) {
        final Mono<Person> dbPerson = getById(id);
        if (Objects.isNull(dbPerson)) {
            return Mono.empty();
        }
        return getById(id).switchIfEmpty(Mono.empty()).filter(Objects::nonNull).flatMap(personToBeDeleted -> personRepository
                .delete(personToBeDeleted).then(Mono.just(personToBeDeleted)));
    }
}

我了解除了save 和update 方法之外的所有内容。我不明白为什么我们在这种情况下使用flatMap。 为什么会这样，如何在我的 Handler 中编写 update 方法的实现。

更新

我们看一下Handler中的save()方法

public Mono<ServerResponse> save(ServerRequest request) {
        final Mono<Person> person = request.bodyToMono(Person.class);
        return ok()
                .contentType(MediaType.APPLICATION_JSON)
                .body(fromPublisher(person.flatMap(personService::save), Person.class));
    }

我认为事实是我们已经收到了：

final Mono<Person> person = request.bodyToMono(Person.class);

然后我们做：

personService::save

结果，我们得到 Mono>

flatMap 就像 map 一样，除了它会解包给定的 lambda 的返回值，如果该值本身包含在 Publisher<T> 中。在我们的例子中，personService.save(T) 方法返回一个Mono<T>。如果我们使用 map 而不是flatMap(T)，我们将有一个Mono< Mono< T>>，而我们真正想要的是Mono<T>。我们可以使用 flatMap 彻底解决这个问题。

我是对的还是这个说法是错的？

Answer 1

为什么需要 flatMap。

这些是我的想法，答案取决于你是在 Mono 还是 Flux 上工作。

1.

方法map和flatMap的javadoc展示了它们的用法：

map：通过对其应用同步函数来转换此 {@link Mono} 发出的项目。

flatMap：异步转换此 {@link Mono} 发出的项目，返回另一个 {@link Mono} 发出的值（可能更改值类型）。

也就是说，将flatMap 和map 视为具有输入和输出的管道，当输出是相同的项目时使用map，否则使用flatMap。检查这个：

public Mono<ServerResponse> influCRUD(ServerRequest req) {
    return req.bodyToMono(S.class) // the pipline begins with S class.
       .map(s -> {s.setF1(f1); s.setF2(f2); return s;}) // the pipeline has the same intput and output, i.e. object s, you use map.
       .flatMap(s -> webClient // the pipeline has S input, and T output, you use flatMap
            .post()
            .uri(uri)
            .body(BodyInserters.fromObject(s))
            .retrive()
            .bodyToMono(T.class) 
        ).flatMap(t -> ServerResponse // now the pipeline changes again, you use flatMap.
           .ok()
           .contentType()
           .body(BodyInserters.fromObject(t))
        );
}

值得一提的是map 也可以有不同的对象作为输出。

1. flatMap 处理每个项目

以上原因对Mono生产者很有用。对于Flux，flatMap 处理所有项目，而map 处理所有项目（或一项）。这与它们在 lambda 中的相同。如果您想处理每个项目，请使用flatMap。

1. flatMap 为你脱掉一层 Mono。

看看他们的声明：

<R> Mono<R> map(Function<? super T, ? extends R> mapper)

和

<R> Mono<R> flatMap(Function<? super T, ? extends Mono<? extends R>> transformer)

Function 除了a -> b 什么都不做，当b 是另一个Producer/Subsciber 的输出时（这很可能在您使用反应式编程时），就像前一个示例中的webClient 部分一样，它在Mono 或 Flux 的形式。通过使用 flatMap，它会为您返回 Mono<R>，其中 map 返回 Mono<Mono<R>>，正如函数声明中所述。

我也是响应式编程的初学者，非常欢迎纠正这个问题。