Project Reactor：用 Mono 的结果丰富 Flux 的结果

Question

我正在努力将反应式编程引入我的公司。我正在构建一个活动推荐系统的简单演示，以证明性能优势。

我遇到的一个挑战是用另一个流的结果来丰富一个流中的结果。我有一个工作示例，如下所示，但我不确定这种方法是否存在任何问题。有人可以看看并提供任何潜在的改进。

    public Flux<Integer> getRecommendedActivities(Long userId, String location, Integer limit) {
        Flux<ActivityData> activities = activityDatabaseService.getByLocation(location);
        Mono<Map<String,BigInteger>> userCategoryScores = userScoresDatabaseService.get(userId);

        return activities
            .zipWith(userCategoryScores.cache().repeat(), this::scoreActivitiesBasedOnUserCategoryScores)
            .sort(compareActivityScoreStrength)
            .map(ScoredActivityData::getActivityId)
            .take(limit);
    }

    private ScoredActivityData scoreActivitiesBasedOnUserCategoryScores(ActivityData deal,Map<String, BigInteger> categoryScores){
        //This method combines the deal score and user category scores to come up with a final score
    }

谢谢， 卡尔

Answer 1

那里的代码本质上没有错。一些可能有用也可能没用的风格点：

• 反应式编程的“规范”是使用流畅的样式并在整个过程中内联所有内容，而不是在方法的顶部声明单独的局部变量并在反应链中使用这些变量。

• x.zipWith(y.cache().repeat()) 模式可以正常工作，但如果可以避免，我觉得它有点难看（zipWith() 意味着我脑海中有两个真实 Flux 数据，而不是Mono 被任意缓存和重复 - 所以行为不一定像它可能的那样“突出”。）相反，我更喜欢 y.flatMapMany(x) - 它让你更清楚地看到一个值并应用通过 real Flux 转换为多个值。所以在你的情况下，这可能看起来像：

  userScoresDatabaseService.get(userId)
          .flatMapMany(c -> activityDatabaseService.getByLocation(location)
                  .map(a -> scoreActivitiesBasedOnUserCategoryScores(a, c))
          )
          .sort() //etc.
  

• Flux.sort() 确实应该是“最后的手段”操作，尤其是因为您将性能优势作为探索响应式的原因。 （从数据库服务中读取整个数据，然后对其进行排序，然后只获取第一个 n 值会导致效率低下 - 你会更好地对数据层中的值进行排序和限制。）请记住 Flux.sort() 必须在排序和返回值之前等待整个源 Flux 完成，将每个值存储在内存中，因此这样做首先会失去Flux 的许多好处地方。它还使您的反应链更短更简单，因为它不需要担心排序和限制。
• Nit：您的某些方法名称和变量名称似乎很长。如果可能的话，我会缩短它们 - 我发现这使得在反应链的上下文中更容易阅读。

根据以上几点，您的整个getRecommendedActivities() 可能会是这样的：

scoresDb.get(userId)
        .flatMapMany(c -> activityDb.getByLocation(location, limit, comparator)
                .map(a -> score(a, c).getActivityId())
        )

...至少对我来说，读起来更短更简单。