1.mongo的聚合管道查询，也就是连表查询，在下面的例子中没有用mongodb的语句查询，使用了封装的类。
public Flux<PostSaleHspAggregateBo> getPostSaleWaitCompletedList(PostSaleSearch search, Pageable pageable) { Criteria criteria1 = new Criteria(); String noGroup = "未分组"; if (StringUtils.isNotBlank(search.getGroupName()) && search.getGroupName().equals(noGroup)) { criteria1.and("hospital.group_name").exists(false); } else if (StringUtils.isNotBlank(search.getGroupName()) && !search.getGroupName().equals(noGroup)) { criteria1.and("hospital.group_name").is(search.getGroupName()); } return reactiveMongoTemplate.aggregate( Aggregation.newAggregation( Aggregation.match(getCriteria(new Criteria(), search)), Aggregation.group("service_organ_code") .count().as("num") .sum(ConditionalOperators.Cond.when(Criteria.where("priority").is(Globals.PostSale.Priority.VeryUrgent.getNum())) .then(1).otherwise(0)).as("countVeryUrgentPriority") .sum(ConditionalOperators.Cond.when(Criteria.where("priority").is(Globals.PostSale.Priority.Urgent.getNum())) .then(1).otherwise(0)).as("countUrgentPriority") .sum(ConditionalOperators.Cond.when(new Criteria().andOperator( Criteria.where("priority").ne(Globals.PostSale.Priority.VeryUrgent.getNum()), Criteria.where("priority").ne(Globals.PostSale.Priority.Urgent.getNum()) )) .then(1).otherwise(0)).as("countNormalPriority") .min("created_at").as("createdAtBegin") .max("created_at").as("createdAtEnd"), //连表查询 trade_post_sale表和hsp_hospital表 根据trade_post_sale表 中organization_code和hsp_hospital表中的医院编码做连接查询，将查询到的医院信息放到hospital上面 Aggregation.lookup("hsp_hospital", "_id", "organization_code", "hospital"), //展开hospital信息 Aggregation.unwind("$hospital"), Aggregation.match(criteria1), //取出你想要的信息字段包含hsp_hospital表中的字段和trade_post_sale表中的字段 Aggregation.project("num", "countVeryUrgentPriority", "countUrgentPriority", "countNormalPriority", "createdAtBegin", "createdAtEnd") .and("_id").as("hspCode") .and("hospital.name").as("hspName") .and("hospital.cooperation_type").as("cooperationType") .and("hospital.support_verify_type").as("supportVerifyType") .and("hospital.settlement_type").as("settlementType") .and("hospital.group_name").as("groupName"), Aggregation.sort(Sort.Direction.DESC, "countVeryUrgentPriority") .and(Sort.Direction.DESC, "countUrgentPriority") .and(Sort.Direction.DESC, "num"), Aggregation.skip(pageable.getOffset()), Aggregation.limit(pageable.getPageSize()) ), "trade_post_sale", PostSaleHspAggregateBo.class ); }2.上面用到的getCriteria方法，主要用于添加搜索条件。
private Criteria getCriteria(Criteria criteria, PostSaleSearch search) { criteria.and("processed").is(false).and("honeypot").is(false); if (StringUtils.isNotBlank(search.getPostSaleType())) { criteria.and("post_sale_type").in(search.getPostSaleType().split(",")); } if (StringUtils.isNotBlank(search.getApplyTimeBegin()) && StringUtils.isNotBlank(search.getApplyTimeEnd())) { criteria.and("created_at").gte(search.getApplyTimeBegin()).lte(search.getApplyTimeEnd()); } else if (StringUtils.isNotBlank(search.getApplyTimeEnd())) { criteria.and("created_at").lte(search.getApplyTimeEnd()); } else if (StringUtils.isNotBlank(search.getApplyTimeBegin())) { criteria.and("created_at").gte(search.getApplyTimeBegin()); } String keyword = search.getKeyword(); if (StringUtils.isNotBlank(keyword)) { Criteria criteria1 = MongoUtil.fuzzySearchCriteria(keyword, "post_sale_no", "service_code", "service_organ_code", "service_organ_name", "user_code", "user_full_name", "user_phone", "customer_full_name", "customer_phone" ); criteria.andOperator(criteria1); } return criteria; }3.统计数量的方法
public Mono<Long> countPostSaleWaitCompletedList(PostSaleSearch search) { Criteria criteria1 = new Criteria(); String noGroup = "未分组"; if (StringUtils.isNotBlank(search.getGroupName()) && search.getGroupName().equals(noGroup)) { criteria1.and("hospital.group_name").exists(false); } else if (StringUtils.isNotBlank(search.getGroupName()) && !search.getGroupName().equals(noGroup)) { criteria1.and("hospital.group_name").is(search.getGroupName()); } return reactiveMongoTemplate.aggregate( Aggregation.newAggregation( Aggregation.match(getCriteria(new Criteria(), search)), Aggregation.group("service_organ_code").count().as("num"), Aggregation.project("num").and("_id").as("hspCode"), Aggregation.lookup("hsp_hospital", "_id", "organization_code", "hospital"), Aggregation.unwind("$hospital"), Aggregation.match(criteria1), Aggregation.count().as("count") ), "trade_post_sale", AggregationCount.class) .next() .defaultIfEmpty(new AggregationCount(0L)) .map(aggregationCount -> aggregationCount.getCount()); }4.最后接口中调用，完整的接口示例。
@GetMapping(Hplus.platform.trade.post_sale_wait_completed) public Mono<HplusResponse<List<PostSaleHspAggregateBo>>> reserveWaitCompletedList(PostSaleSearch search, @RequestParam(required = false, defaultValue = "10") Integer pageSize, @RequestParam(required = false, defaultValue = "1") Integer pageNo) { return postSaleService.getPostSaleWaitCompletedList(search, PageRequest.of(pageNo-1 , pageSize)) .collectList() .zipWith(postSaleService.countPostSaleWaitCompletedList(search)) .flatMap(HplusResponse::ok); }