mongoDB聚合查询实现

mongo简介

百度百科上的介绍：MongoDB是一个基于分布式文件存储的数据库。由C++语言编写。旨在为WEB应用提供可扩展的高性能数据存储解决方案。MongoDB 将数据存储为一个文档，数据结构由键(key=>value)对组成。MongoDB 文档类似于 JSON 对象。字段值可以包含其他文档，数组及文档数组。

为啥用捏

其实还是因为之前做的一个功能，涉及到的数据量比较大，存在数据库中用sql查询执行起来很慢很慢，优化之后还是很慢。具体可以参考我写的那篇sql优化的文章，是的~大概半年前的一个功能了，由于优先级是最低的，所以到现在才去优化和总结，我是有够懒的啦。不过最后还是做了，成功将数据从pg转到mongo了，在实现复杂查询的时候也是费了一下小劲呢，所以总结下来供以后来回顾一下。

正题

之前的业务数据存入数据库之后还要关联其他的表去做一个关联查询，由于也没打算把那些数据再放mongo里面一份，所以设计数据结构的时候也就把关联的字段在代码中处理了一下全都放到一个json里面了。差不多就那样子，这样就不需要设计联表查询，只需要对这个collection进行操作就行啦。

{ 
    "_id" : NumberInt(2), 
    "id" : NumberInt(2), 
    "cameraName" : "测试点", //设备名称
    "clientIp" : "0.0.0.0", 
    "endTime" : NumberLong(20191217102949), 
    "status" : NumberInt(2), //取流状态  0：正在取流 2 取流结束 1  取流异常
    "retrivalDuration" : NumberLong(200), 
    "createTime" : "2020-02-18 16:46:27", 
    "provinceCode" : "360112", //行政区划编码
    "eurl" : "rtsp://10.5.160.103:655/Eurl/qEPs1Km", //短连接
    "longurls" :"rtsp://10.17.10.21:655/0", 
    "clientType" : "BROWSER", //客户端类型
    "cascadeType" : NumberInt(0),  //("所属域 0:本级  1:下级  2：上级 3：中间") 
    "userId" : "test", 
    "playType" : "playback", //playback ,realplay 播放类型
    "startTime" : NumberLong(20191217102749), 
    "upCascadeIndexCode" : "36010000002000000088", //上级域编号，
    "downCascadeIndexCode" : "36010000002000000088", //下级域编号
    "cameraIndexCode" : "36011240201310084354", //点位编号
    "sessionId" : "21312xada", //会话id
    "mediaIndexCode" : "3601124020131002321", //媒体编号
    "msgType" : NumberInt(0), //消息类型 0:正常链路上报 1：媒体的定时数据校正 2：媒体重启上报
    "path" : "@100000@360000@360100@360112@", //行政区划全路径
    "placeName" : ""//行政区划名称
}

聚合查询

这里面涉及到什么操作呢，分页，模糊查询，求和函数，分组操作，区间查询，然后把这些拼到一起~~

一开始用的时候由于对mongo并没有很了解，不知道mongo聚合操作其实是有管道的，代码里顺序也没有很注意，导致代码执行之后没有报错但是查询结果一直不对。这里简单说一下管道，也就是aggregation pipeline。

MongoDB的聚合管道将MongoDB文档在一个管道处理完毕后将结果传递给下一个管道处理。管道操作是可以重复的。

表达式：处理输入文档并输出。表达式是无状态的，只能用于计算当前聚合管道的文档，不能处理其它的文档。

这里我们介绍一下聚合框架中常用的几个操作：

$project：修改输入文档的结构。可以用来重命名、增加或删除域，也可以用于创建计算结果以及嵌套文档。

$match：用于过滤数据，只输出符合条件的文档。match使用MongoDB的标准查询操作。

$limit：用来限制MongoDB聚合管道返回的文档数。

$skip：在聚合管道中跳过指定数量的文档，并返回余下的文档。

$unwind：将文档中的某一个数组类型字段拆分成多条，每条包含数组中的一个值。

$group：将集合中的文档分组，可用于统计结果。

$sort：将输入文档排序后输出。

$geoNear：输出接近某一地理位置的有序文档。

本次查询就用到了上面大部分的操作，具体看下面代码吧：

		Integer page =  resourceRetrivalSearchParamsVO.getPage();
        Integer pageSize = resourceRetrivalSearchParamsVO.getRows();
        Long skip = (page-1) * pageSize.longValue();
        //用来构建条件
        Criteria criteria = new Criteria();
        List<AggregationOperation> operations = new ArrayList<>();
        String name = resourceRetrivalSearchParamsVO.getName();
        String path = resourceRetrivalSearchParamsVO.getProvinceCode();
        if(!StringUtils.isEmpty(name)){
            Pattern namePattern = Pattern.compile(DefaultEnum.CHARACTER_SPOT_STAR.getValue() + name + DefaultEnum.CHARACTER_SPOT_$.getValue(), Pattern.CASE_INSENSITIVE);
            operations.add(Aggregation.match(criteria.where("cameraName").regex(namePattern)));
        }
        if(!StringUtils.isEmpty(path)){
            Pattern pathPattern = Pattern.compile(DefaultEnum.CHARACTER_SPOT_STAR.getValue() + path + DefaultEnum.CHARACTER_SPOT_$.getValue(), Pattern.CASE_INSENSITIVE);
            operations.add(Aggregation.match(criteria.where("path").regex(pathPattern)));
        }

        if(!StringUtils.isEmpty(resourceRetrivalSearchParamsVO.getStartTime())
                &&  !StringUtils.isEmpty(resourceRetrivalSearchParamsVO.getEndTime())){
            //大于方法
            Criteria gt = criteria.where("updateTime").gte(resourceRetrivalSearchParamsVO.getStartTime());
            //小于方法
            Criteria lt = criteria.where("updateTime").lte(resourceRetrivalSearchParamsVO.getEndTime());
            operations.add(Aggregation.match(gt));
            operations.add(Aggregation.match(lt));
        }
        operations.add(Aggregation.group("cameraIndexCode","provinceCode","cameraName","placeName").
                sum("retrivalDuration").as("retrievalDuration").count().as("retrievalNum"));
        operations.add(Aggregation.project("cameraIndexCode","provinceCode","cameraName","placeName","retrievalDuration","retrievalNum"));
        operations.add(Aggregation.sort(Sort.Direction.ASC,"cameraIndexCode"));
        Integer total = mongoTemplate.aggregate(Aggregation.newAggregation(operations),MONGO_COLLECTION_NAME,Map.class)
                .getMappedResults().size();
        operations.add(Aggregation.skip(skip));
        operations.add(Aggregation.limit(pageSize.longValue()));
        Aggregation aggregation = Aggregation.newAggregation(operations);
        List<Map> results = mongoTemplate.aggregate(aggregation,MONGO_COLLECTION_NAME, Map.class).getMappedResults();

具体写的够不够简洁或者够不够优化咱也不知道，但是是可以按照我的需求查询出数据的。其实拼接这些条件的时候在代码里是很抽象的。这里其实可以合理是用工具的，studio 3T真的是个很好用的工具了，里面的功能也是很完善，其实对这种拼接条件不是很熟练的话可以把sql写好，然后去里面执行一下转换成聚合查询语句的，这样你就知道你需要按什么顺序去拼接这些条件了，具体如图，这样效率还是很高的，而且你可以在debug调试的时候把你拼接的查询语句粘贴出来去工具中一步一步的验证一下。

// Requires official MongoShell 3.6+
use local;
db.getCollection("ipdirmgr_resource_retrival_info").aggregate(
    [
        { 
            "$match" : { 
                "start_time" : { 
                    "$gte" : ""
                }, 
                "end_time" : { 
                    "$lte" : ""
                }, 
                "cameraName" : /^\x{6d4b}$/i, 
                "path" : /^$/i
            }
        }, 
        { 
            "$group" : { 
                "_id" : { 
                    "cameraIndexCode" : "$cameraIndexCode", 
                    "cameraName" : "$cameraName", 
                    "provinceCode" : "$provinceCode", 
                    "placeName" : "$placeName"
                }, 
                "SUM(retrivalDuration)" : { 
                    "$sum" : "$retrivalDuration"
                }, 
                "COUNT(cameraIndexCode)" : { 
                    "$sum" : NumberInt(1)
                }
            }
        }, 
        { 
            "$project" : { 
                "cameraIndexCode" : "$_id.cameraIndexCode", 
                "provinceCode" : "$_id.provinceCode", 
                "cameraName" : "$_id.cameraName", 
                "placeName" : "$_id.placeName", 
                "SUM(retrivalDuration)" : "$SUM(retrivalDuration)", 
                "COUNT(cameraIndexCode)" : "$COUNT(cameraIndexCode)", 
                "_id" : NumberInt(0)
            }
        }, 
        { 
            "$skip" : NumberInt(1)
        }, 
        { 
            "$limit" : NumberInt(25)
        }
    ], 
    { 
        "allowDiskUse" : true
    }
);

总结

总之这篇还是偏实用性吧，对mongo具体的原理或者更细化的介绍并没有，简单的插入更新之类的操作就不说了，用mongoTemplate实现很简单。其他的等我先去好好学习一下，后面再来总结吧~