Spring Reactive 实现多次调用 db答案

【问题标题】：Spring Reactive implementation calling db multiple timesSpring Reactive 实现多次调用 db
【发布时间】：2020-06-28 17:16:43
【问题描述】：

我是被动的新手，并试图以有效的方式完成以下任务。我有一个表，每个用户都有事件。我正在尝试获取使用最新类别过滤的给定用户的每个事件名称的最新行。

table structure
user id , category , event name, details , insert timestamp, event text( payload of event)

反应式方法 // 调用 1 到 Cassandra

Mono<Event> latestCategory = repository.findByUserId(userId).sort().next(); // sort is by insert 
timestamp;

//调用 2 到 Cassandra

Flux<Event> fluxEvents = repository.findByUserId(userId)
                .groupBy(Event::name) //grouping by event name
                 .flatmap(grp -> {
                 grp.sort() // sorting for each event
                 grp.next().zipWith(latestCategory) // picking latest row for each event
                 .filter(eventWithLatestCategory -> 
eventWithLatestCategory.getT1().category.equals(eventWithLatestCategory.getT2().category) //filtering by each category
.map(Tuple2::getT1)// picking latest event row for latest category
};

功能上一切正常，但问题是我看到表中的每一行都发生了 DB 调用。在命令式编程中，我可以通过一个 db 调用来实现它，然后应用上述逻辑。我怎样才能在反应式世界中做同样的事情？

enter code here

【问题讨论】：

标签： reactive-programming project-reactor reactive imperative-programming

【解决方案1】：

“最轻松”的方法就是你要做的事情：

Mono<Event> latestCategory = repository.findByUserId(userId).sort().next().cache();

...这意味着 latestCategory 只会被获取一次，然后缓存以供所有后续订阅使用。

但这可能不是最佳解决方案。

在这种情况下，就目前而言，您正在Flux 本身中进行排序。这通常不明智，但如果您绝对需要这样做，那么您也可以这样做：

repository.findByUserId(userId).sort().collectList().map(eventList -> {
    //Deal imperatively with a List<Event>
});

...然后可以通过单个数据库调用访问列表，您可以根据需要随机查询。您实际上是在获得，而不会以这种方式损失性能，因为只有一个数据库调用 - 并且作为排序的 Flux 在整个发布者完成之前永远不会输出任何内容，这与简单地收集到一个列表中。

然而，更好的方法是让 cassandra 进行底层排序，然后使用 switchOnFirst() 将第一个元素（您的最新类别）与其他所有元素一起压缩：

repository.findByUserId(userId).switchOnFirst((signal, flux) ->
    flux.map(val -> Tuples.of(signal.get(), val)) //(In real-world use, check the `flux` actually has a value first)
)
//...etc

这意味着：

您只需要一个数据库查询；
您可以在每个值出现时对其进行反应性处理，而无需等待整个流结束；
无需缓存发布者，如有必要，最好避免。

【讨论】：