Spring WebFlux + MongoDB：可尾游标和聚合答案

【问题标题】：Spring WebFlux + MongoDB: Tailable Cursor and AggregationSpring WebFlux + MongoDB：可尾游标和聚合
【发布时间】：2020-08-10 13:18:43
【问题描述】：

我是 WebFlux 和 MongoDB 的新手。我正在尝试在带有可尾光标的上限集合中使用聚合，但我没有成功。我想执行这个 mongoDB 查询：

db.structures.aggregate(
   [
     {
        $match: {
            id: { $in: [8244, 8052]}
        }    
     },  
     { $sort: { id: 1, lastUpdate: 1} },
     {
       $group:
         {
           _id: {id: "$id"},
           lastUpdate: { $last: "$lastUpdate" }
         }
     }
   ]
)

ReactiveMongoOperations 让我可以选择“tail”或“aggregation”。

我能够执行聚合：

    MatchOperation match = new MatchOperation(Criteria.where("id").in(8244, 8052));
    GroupOperation group = Aggregation.group("id", "$id").last("$lastUpdate").as("lastUpdate");
    Aggregation aggregate = Aggregation.newAggregation(match, group);

    Flux<Structure> result = mongoOperation.aggregate(aggregate,
            "structures", Structure.class);

或尾光标

    Query query = new Query();
    query.addCriteria(Criteria.where("id").in(8244, 8052));
    Flux<Structure> result = mongoOperation.tail(query, Structure.class);

有可能吗？尾部和聚合在一起？

使用聚合是我发现只为每个 id 获取最后插入的文档的方式。

没有聚合我得到：

query without aggregation

有聚合：

query with aggregation

提前通知

【问题讨论】：

标签： mongodb aggregation-framework spring-webflux capped-collections

【解决方案1】：

tailable 游标查询创建一个永远不会完成的Flux（永远不会发出onComplete 事件），并且Flux 在插入数据库时发出记录。由于这个事实，我认为数据库引擎不允许使用可尾游标进行聚合。

所以聚合在某种程度上没有意义，因为在每条新插入的记录上都需要重新计算聚合。从技术上讲，您可以进行运行聚合，为每个返回的记录计算所需的聚合记录并将其发送到下游。

一种可能的解决方案是以编程方式对返回的“无限”Flux 进行聚合：

mongoOperation.tail(query, Structure.class)
  .groupBy(Structure::id) // create independent Fluxes based on id
  .flatMap(groupedFlux -> 
    groupedFlux.scan((result, nextStructure) -> { // scan is like reduce but emits intermediate results
    log.info("intermediate result is: {}", result);
    if (result.getLastUpdate() > nextStructure.getLastUpdate()) {
        return result;
    } else {
        result.setLastUpdate(nextStructure.getLastUpdate());
        return result;
    }
}));

另一方面，您可能应该重新审视您的用例以及您需要在此处完成的工作，看看是否应该使用上限集合以外的其他内容，或者聚合部分可能是多余的（即，如果新插入的记录始终具有 @987654326 @ 属性大于上一条记录）。

【讨论】：