springboot Mongo大数据查询优化方案

川长思鸟来 2024-04-22 15:28 82阅读 0赞

## 前言 ##

因为项目需要把传感器的数据保存起来，当时设计的时是mongo来存储，后期需要从mongo DB里查询传感器的数据记录。由于传感器每秒都会像mongo数据库存500条左右的数据，1天就有4320万条数据，要想按照时间条件去查询，经常会被卡死。以下是我的解决过程和方案。

## 解决方案 ##

### 水平分表 ###

#### 按照传感器类型分表 ####

将不同不同传感器的数据，分别存入不同的表（集合）中，这样每个表的数据就成倍减少，但是过了一段时间发现查询嗨是很慢，每个传感器每秒需要保存的数据也有100条左右，一天就是864万条数据。仅靠类型分表是不行的。

#### 按照日期分表 ####

每个表每天的数据有864万条数据，一个月就是2.6亿条数据。于是按照日期，每天对每个传感器类型建设了一个表 表（集合）名格式如 ‘temperature\_sensor\_20240310’。  
如保存数据时候自动创建分表代码如下：

*  DateUtil.today() 是我工具类里的方法等效于 DateUtil.format(new Date(),“yyyyMMdd”)
 *  注意请保证每个传入的对象里都有一个createTime字段，查询的时候会用到

按照时间查询分表的方法，代码如下：

public <T> List<T>  getSecondData(LocalDateTime start, LocalDateTime end, Class<T> entityClass, String collectionName) {
        
            String collectionNameSharding =collectionName+"_"+DateUtil.format(start,"yyyyMMdd");
            // 设置时间范围查询条件
            Criteria criteria = Criteria.where("createTime")
                    .gte(start)
                    .lte(end);
            // 查询数据
            return mongoTemplate.find(Query.query(criteria).limit(1000).skip(0), entityClass,collectionNameSharding);
        }

*  代码中的 **.limit(1000)** 表示限制查询结果的数量，即最多返回1000条匹配的文档记录。这对于分页查询或者批量处理数据时非常有用，可以避免一次性加载过多数据导致内存溢出或响应延迟。
 *  而 **.skip(0)** 则表示跳过前0条匹配的文档记录，从第一条开始返回。在分页查询场景下，如果你想获取第二页的数据，通常会将skip的参数设置为每页大小（假设也是1000），即 .skip(1000)，这样就会跳过前1000条，然后取接下来的1000条数据。
    
    经过以上操作查询数据的时候不会被卡顿了，但是查询速度需要2s左右,项目需求查询速度至少得在200ms内，所以还得继续优化。

### 建立索引 ###

因为mongo水平分表的缘故，不可能人工去对每个字段创建的表（集合）去建立时间索引，需要代码实现，创建表的同时，自动创建时间索引。

*  修改分表数据保存方法如下：

@Async
        public <T> void insertSharding(Collection<? extends T> batchToSave, String collectionName) {
        
            String collectionNameSharding = collectionName + "_" + DateUtil.today();
            if (!mongoTemplate.collectionExists(collectionNameSharding)) {
        
                mongoTemplate.createCollection(collectionNameSharding);
                IndexOperations indexOps = mongoTemplate.indexOps(collectionNameSharding);
                indexOps.ensureIndex(new Index().on("createTime", Sort.Direction.ASC).named(collectionNameSharding+"_createTime"));
            }
            if (CollectionUtil.isNotEmpty(batchToSave)) {
        
                mongoTemplate.insert(batchToSave, collectionNameSharding);
            }
        }

*  named(collectionNameSharding+“\_createTime”)) 即创建索引的名称
 *  on(“createTime”, Sort.Direction.ASC) 即使用集合中的createTime字段按照升序建立索引。

## 总结 ##

经过以上水平分表和建立索引的方法，按照时间条件去查询的方法已经可以优化到200ms左右了。本篇教程到此未知，如果觉得不错，记得一键三连，感谢各位的支持！！！