一 : 需求: 根据sn码值进行分组,获取每个分 组中成功个数,进行计算成功率.

代码:

 /**
     * 音箱播报成功率监控 分页查询
     *
     * @param soundBoxBroadcastQuery 音箱播报入参查询
     * @return {@link R}
     */
   @GetMapping("soundBoxBroadcastMonitor")
    public R soundBoxBroadcastMonitor(@Valid SoundBoxBroadcastQuery soundBoxBroadcastQuery) {
        logger.debug("音箱播报成功率监控入参：{}", soundBoxBroadcastQuery.toString());
        BoolQueryBuilder boolQueryBuilder = getBoolQueryBuilder(soundBoxBroadcastQuery);

        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        queryBuilder.withQuery(boolQueryBuilder);
        queryBuilder.withSort(SortBuilders.fieldSort("_id").order(SortOrder.DESC));
        queryBuilder.withPageable(PageRequest.of(
                Integer.valueOf(soundBoxBroadcastQuery.getPageNum()) - 1,
                Integer.valueOf(soundBoxBroadcastQuery.getPageSize())));

        // 根据音响SN码分类
        TermsAggregationBuilder aggregationBuilder = AggregationBuilders.terms("group_by_equipmentCode").field("equipmentCode").size(99999999);
        queryBuilder.addAggregation(aggregationBuilder);
        logger.info("查询条件：{}", boolQueryBuilder.toString());
        List<SoundBoxBroadcastVO> list = new ArrayList<>();
        try {
            Page<BroadcastMonitor> items = this.broadcastMonitorRepository.search(queryBuilder.build());
            Aggregations aggregations = elasticsearchRestTemplate.query(queryBuilder.build(), response -> response.getAggregations());
            list = getAggregationsResult(items, aggregations);
        } catch (Exception e) {
            logger.error("音箱播报成功率监控", e);
        }
        logger.debug("音箱播报成功率监控出参：{}", list);
        return R.ok().put("data", list).put("count", list.size());
    }


/**
     * 封装查询结果
     *
     * @param items 查询列表参数
     * @param aggregations 聚合
     * @return {@link List}<{@link SoundBoxBroadcastVO}>
     */
    private List<SoundBoxBroadcastVO> getAggregationsResult(Page<BroadcastMonitor> items, Aggregations aggregations) {
        List<SoundBoxBroadcastVO> list = new ArrayList<>();
        Map<String, Aggregation> stringAggregationMap = aggregations.asMap();
        ParsedTerms group_by_equipmentCode = (ParsedTerms) stringAggregationMap.get("group_by_equipmentCode");
        List<? extends Terms.Bucket> buckets = group_by_equipmentCode.getBuckets();
        for (Terms.Bucket bucket : buckets) {
            SoundBoxBroadcastVO soundBoxBroadcastVO = new SoundBoxBroadcastVO();
            soundBoxBroadcastVO.setEquipmentCode(bucket.getKeyAsString());
            soundBoxBroadcastVO.setAcceptMessageCount(bucket.getDocCount());
            long count = items.stream().filter(m -> m.getEquipmentCode().equals(bucket.getKeyAsString()) && BROADCAST_SUCCESS.equals(m.getBroadcastStatus())).count();
            DecimalFormat df = new DecimalFormat("0.00");
            soundBoxBroadcastVO.setSuccessMessagePercent(df.format(Double.valueOf(count) / Double.valueOf(bucket.getDocCount()) * 100) + "%");
            soundBoxBroadcastVO.setSuccessMessageCount(count);
            items.stream().forEach(item -> {
                if (item.getEquipmentCode().equals(bucket.getKeyAsString())) {
                    soundBoxBroadcastVO.setMerId(item.getMerId());
                    soundBoxBroadcastVO.setUnionPayId(item.getUnionPayId());
                    soundBoxBroadcastVO.setBroadcastType(item.getBroadcastType());
                }
            });
            list.add(soundBoxBroadcastVO);
        }
        return list;
    }

另外一种思路,根据子聚合分组查询

/*  *//**
     * 音箱播报成功率监控 分页查询
     *
     * @param soundBoxBroadcastQuery 音箱播报入参查询
     * @return {@link R}
     *//*
    @GetMapping("soundBoxBroadcastMonitor")
    public R soundBoxBroadcastMonitor(@Valid SoundBoxBroadcastQuery soundBoxBroadcastQuery) {
        logger.debug("音箱播报成功率监控入参：{}", soundBoxBroadcastQuery.toString());
        BoolQueryBuilder boolQueryBuilder = getBoolQueryBuilder(soundBoxBroadcastQuery);

        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        queryBuilder.withQuery(boolQueryBuilder);
        queryBuilder.withSort(SortBuilders.fieldSort("_id").order(SortOrder.DESC));
        queryBuilder.withPageable(PageRequest.of(
                Integer.valueOf(soundBoxBroadcastQuery.getPageNum()) - 1,
                Integer.valueOf(soundBoxBroadcastQuery.getPageSize())));

        // 根据音响SN码分类
        TermsAggregationBuilder aggregationBuilder = AggregationBuilders.terms("group_by_equipmentCode").field("equipmentCode").size(999999)
                .subAggregation(AggregationBuilders.terms("group_by_broadcastStatus").field("broadcastStatus"));

        queryBuilder.addAggregation(aggregationBuilder);

        logger.info("查询条件：{}", boolQueryBuilder.toString());
        List<SoundBoxBroadcastVO> soundBoxBroadcastVOList = null;
        try {
            Page<BroadcastMonitor> items = this.broadcastMonitorRepository.search(queryBuilder.build());
            Aggregations aggregations = elasticsearchRestTemplate.query(queryBuilder.build(), response -> response.getAggregations());

            *//*List<SoundBoxBroadcastVO> list = new ArrayList<>();
            Map<String, Aggregation> stringAggregationMap = aggregations.asMap();
            ParsedTerms group_by_equipmentCode = (ParsedTerms) stringAggregationMap.get("group_by_equipmentCode");
            List<? extends Terms.Bucket> buckets = group_by_equipmentCode.getBuckets();
            long docCount = buckets.size();
            for (Terms.Bucket bucket : buckets) {
                SoundBoxBroadcastVO soundBoxBroadcastVO = new SoundBoxBroadcastVO();
                soundBoxBroadcastVO.setEquipmentCode(bucket.getKeyAsString());
                soundBoxBroadcastVO.setAcceptMessageCount(bucket.getDocCount());
                Terms group_by_broadcastStatus = (Terms) bucket.getAggregations().get("group_by_broadcastStatus");
                for (Terms.Bucket bucket1 : group_by_broadcastStatus.getBuckets()) {
                    if (BROADCAST_SUCCESS.equals(bucket1.getKeyAsString())) {
                        DecimalFormat df = new DecimalFormat("0.00");
                        soundBoxBroadcastVO.setSuccessMessagePercent(df.format(Double.valueOf(bucket1.getDocCount()) /Double.valueOf(bucket.getDocCount())));
                        soundBoxBroadcastVO.setSuccessMessageCount(bucket1.getDocCount());
                    }
                }
                items.stream().forEach(item->{if(item.getEquipmentCode().equals(bucket.getKeyAsString())){
                    soundBoxBroadcastVO.setMerId(item.getMerId());
                    soundBoxBroadcastVO.setUnionPayId(item.getUnionPayId());
                    soundBoxBroadcastVO.setBroadcastType(item.getBroadcastType());
                }});
                list.add(soundBoxBroadcastVO);
            }
            *//*
            soundBoxBroadcastVOList = getAggregationsResult(items, aggregations);
        } catch (Exception e) {
            logger.error("音箱播报成功率监控", e);
        }
        logger.debug("音箱播报成功率监控出参：{}", soundBoxBroadcastVOList);
        return R.ok().put("data", soundBoxBroadcastVOList).put("count", soundBoxBroadcastVOList.size());
    }*/