elasticsearch - 以 react 方式使用 Spring 数据 Elasticsearch 保存记录通量

Question

请帮助我找到正确的方法来保存从 Cassandra 接收的实体流量到 Elasticserch。
例如，我有大量的实体:

Flux<User> user = cassandraService.getUsers();

如何使用响应式方式将所有实体流量保存到 Elasticsearch？
调用Elasticksearch哪种方法比较好用:

user.map(u-> reactiveElasticsearchOperations.save(u, INDEX_NAME));

user.flatMap(u-> reactiveElasticsearchOperations.save(u, INDEX_NAME));

user.onNext(u-> reactiveElasticsearchOperations.save(u, INDEX_NAME));

……别的……
??

以及如何阻止直到最后一个流动实体被处理？

Answer 1

一、简述flatMap()对比 map() :

flatMap()应该用于非阻塞操作(异步)——简而言之，任何返回 Mono、Flux、

的操作

map()应该用于通过对每个项目应用同步函数来转换对象(在固定时间执行)。

所以如果你想在 Elasticsearch 中保存一个文档，你应该使用 flatMap() - 这是一个异步操作，它的执行时间是不确定的。
说到 doOnNext() , 它用于添加副作用行为，在 Flux 发射项目时触发，例如用于记录或保存完成后应该发生的其他操作。
示例:

Mono.just(productDto)
    .map(dto -> new Product(dto.getName(), dto.isAvailable()))
    .flatMap(elasticsearchOperations::save)
    .doOnNext(savedProduct -> logger.info("Saved {}", savedProduct.getProductName()))
    .subscribe();

而且您不必阻止 save() ，这是 flatMap()的责任.