springboot Mongo大数据查询优化方案

前言

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

解决方案

水平分表

按照传感器类型分表

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

按照日期分表

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

    @Asyncpublic <T> void insertSharding(Collection<? extends T> batchToSave, String collectionName) {String collectionNameSharding = collectionName + "_" + DateUtil.today();if (CollectionUtil.isNotEmpty(batchToSave)) {mongoTemplate.insert(batchToSave, collectionNameSharding);}}

• 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水平分表的缘故，不可能人工去对每个字段创建的表（集合）去建立时间索引，需要代码实现，创建表的同时，自动创建时间索引。

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

  @Asyncpublic <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左右了。本篇教程到此未知，如果觉得不错，记得一键三连，感谢各位的支持！！！