如何解决mongodb深分页的问题

这篇文章主要讲解了“如何解决mongodb深分页的问题”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“如何解决mongodb深分页的问题”吧！

        突然有一天，有用户反馈，保存的个人模板全都不见了，当听到这个消息的时候，第一个想法就是怀疑是用户自己删除了，因为有日常的开发任务，当时并没有在意，自此每隔三五天就有用户反馈同类的问题，这时候下意识的想，之前用户没反馈，现在反馈的多了起来，是不是最近上线的程序有bug，做了删除操作，紧急的查看了一下代码上线记录，发现并没有进行删除模板操作，同时反馈的还有模板加载响应很慢，偶尔接口返回500，测试同事试了一下，加载数据是正常的，抱着质疑的态度开始深入了分析了这部分业务逻辑和程序的编写，发现了一些端倪：

1.     新用户另存为我的模板会创建一个 “个人模板” 的场景，场景id存储在用户的数据表中；

2.     保存的模板页和场景页存储在mongodb的同一个表中，数据表的体量有十亿多条数据；

3.     查询功能有分页，mongodb数据比较大时，对于深分页性能相对较差，分页代码 

   query = new Query(Criteria.where("sceneId").is(sceneId)).with(new Sort(Sort.Direction.DESC, "id"))
   				.skip((page.getPageNo() - 1) * page.getPageSize()).limit(page.getPageSize());

   
   

       查询方式如下图：

 那么显而易见的模板丢失的原因就分析出来了，是用户把 “个人模板” 场景给删除了，从而导致场景下的模板页随之被删除，分析出原因之后，想到了以下的优化方案:

1.     场景列表查询不显示 “个人模板” 场景数据；

2.     优化mongodb分页，优化mongodb分页，优化如下

   if (tplId == null){
      // 分页获取
      query = new Query(Criteria.where("sceneId").is(sceneId)).with(new Sort(Sort.Direction.DESC, "id"))
            .skip((page.getPageNo() - 1) * page.getPageSize()).limit(page.getPageSize());
   }else{
      query = new Query(Criteria.where("sceneId").is(sceneId).lte(tplId)).with(new Sort(Sort.Direction.DESC, "id")).limit(page.getPageSize());
   }

   
   

        优化方案定好之后，很快程序就在预发布部署测试并上线，果不其然，用户在列表看不到 “个人模板” 场景之后，反馈模板丢失的问题没了，但是 个人模板列表加载无响应的问题还在持续存在，难道是mongodb分页优化没起作用？其实并不是，因为表的数据体量已经达到十亿级别，接口的响应最大时间设置为2秒，超过了最大响应时间就返回500，那么mongodb分页优化并不能从根本解决加载慢的问题，新的优化方案随即产生：

1. 场景页和模板页分开存储，统计了一下，模板页的总数一共七百多万，其余的都是场景页数据;

2. 新用户另存为个人模板不产生 “个人模板” 场景，用户表中存储场景ID；

3. 在MySQL库中建立userId和pageId的关系表；

     优化后的查询如下图：

        那么有两个问题，这么大的体量数据，怎么迁移模板页数据呢？     新产生的模板页数据怎么进行存储呢？ 想了一下解决方案：

1.     模板页数据双写，新插入的存储在不同的mongodb表中，并在MySQL中建立 UID和Tpl的关系；

2.      开发模板迁移程序，从用户角度出发，轮训每一位有模板标识的用户获取sceneId，查询出模板页，随之保存到新表，建立UID和Tpl的关系

         接下来就开始去按照这个优化方案去执行：

1.  第一步：在MySQL中建立UID和Tpl的关系表，先进行数据双写 ，数据关系表如下图

2、 第二步，开发模板迁移程序，迁移的办法有好几种，第一种：使用ETL工具进行数据迁移、   第二种：查出历史数据，发送到MQ中，设置一定数量的消费者使用多线程方式去消费执行，最终我觉得最优方案是第二种，如下图：

        流程定义好了，为了不影响业务的正常执行，一般迁移数据这样子的工作都是从数据库的从库获取数据， 接下来就开发迁移程序，首先建立两个项目，data-provider，data-consumer，data-provider 查询用户，把另存为模板的场景ID发送到mq，data-consumer接受场景ID，去查询page，并分别保存到模板页新表和MySQL库的UID和Tpl的关系表中

data-provider代码如下：

@Component
public class TaskTplSyncRunner implements ApplicationRunner {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    private TaskService taskService;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        AtomicInteger total = new AtomicInteger(0);
        AtomicInteger count = new AtomicInteger(1);
        // 开始sceneId
        Long start = null;
        if (!CollectionUtils.isEmpty(args.getOptionValues("start"))) {
            start = args.getOptionValues("start").get(0) == null ? 1 :
                    Long.valueOf(args.getOptionValues("start").get(0));
        }
        // 最后sceneId
        Long end = null;
        if (!CollectionUtils.isEmpty(args.getOptionValues("end"))) {
            end = args.getOptionValues("end").get(0) == null ? 1000 :
                    Long.valueOf(args.getOptionValues("end").get(0));
        }
        // 每一次的执行跨度
        Integer pageSize = null;
        if (!CollectionUtils.isEmpty(args.getOptionValues("pageSize"))) {
            pageSize = args.getOptionValues("pageSize").get(0) == null ? 2000 :
                    Integer.valueOf(args.getOptionValues("pageSize").get(0));
        }
        logger.info("init start value is ={},end value is={}", start, end);
        while (true) {
            Map<String, Object> objectMap = taskService.sendTplMq(start, end);
            if (objectMap.containsKey("endSceneId")){
                // 得到下一次循环的最后一个id
                end = Long.valueOf(objectMap.get("endSceneId").toString());
                start = end - pageSize;
            }
            count.getAndIncrement();
            if (objectMap.containsKey("total")) {
                total.addAndGet(Integer.valueOf(objectMap.get("total") + ""));
            }
            // 是最后一个用户直接跳出
            if (start < 1101) {
                break;
            }
       }
        logger.info("execute personage tpl sync success,total count {} ", total.intValue());
        logger.info("execute personage tpl sync task end。。。。。。。。。。");
    }
}

@Async
    public Map<String, Object> sendTplMq(Long start, Long end) {
        Map<String,Object> paramMap = new HashMap<>();
        paramMap.put("start",start);
        paramMap.put("end",end);
        List<Scene> sceneList = sceneDao.findSceneList(paramMap);

        Map<String,Object> resultMap = new HashMap<>();
        sceneList.stream().forEach(scene -> {
            amqpTemplate.convertAndSend("exchange.sync.tpl","scene.tpl.data.sync.test", JsonMapper.getInstance().toJson(scene));
        });
        resultMap.put("total",sceneList.size());
        resultMap.put("resultFlag",false);
        resultMap.put("endSceneId",start);
        return resultMap;
    }

从ApplicationArguments中获取start，end，pageSize的值的原因是防止程序执行中断，自己设置 VM options

data-consumer 代码如下:

@Component
public class Receiver {

    Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private ScenePageDoc scenePageDoc;
    @RabbitListener(queues = "queue.scene.tpl.data.sync.test", containerFactory = "containerFactory")
    public void receiveTplMessage(String message) {
        Long pageId = null;
        try {
            HashMap<String, Object> hashMap = JsonMapper.getInstance().fromJson(message, HashMap.class);
            pageId = Long.valueOf(String.valueOf(hashMap.get("pageId")));
            // 查询scene_page表中的page信息
            ScenePage scenePage = scenePageDoc.findPageById(pageId);
            if (scenePage != null){
                //查询是否已经同步过
                ScenePageTpl scenePageTpl = scenePageDoc.findPageTplById(pageId);
                if (scenePageTpl == null){
                    scenePageDoc.savePageTpl(scenePage);
                    logger.info("execute sync success pageId value is={}",pageId);
                    // 建立UID 和 tpl 的关系
                }
                // 删除eqs_scene_page表的页面数据
                scenePageDoc.removeScenePageById(pageId);
            }
        }catch (Exception e){
            logger.error("执行同步程序出现异常,error param is ={}", message);
            scenePageDoc.saveSyncError(pageId);
            e.printStackTrace();
        }
    }
}

mq优化

@Configuration
public class MqConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory=new SimpleRabbitListenerContainerFactory();
        // 设置线程池
        ExecutorService service = new ThreadPoolExecutor(60,90,60, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new ThreadPoolExecutor.CallerRunsPolicy());
        factory.setTaskExecutor(service);
        //设置consumer个数
        factory.setConcurrentConsumers(60);
        // 关闭ack
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        configurer.configure(factory,connectionFactory);
        return factory;
    }
}

下一步部署程序

nohup java -jar -Djava.security.egd=file:/dev/./urandom eqxiu-data-provider-0.0.5.jar --start=81947540 --end=81950540 --pageSize=3000 > /data/logs/tomcat/data-provider/spring.log &


nohup java -jar eqxiu-data-consumer-0.0.5.jar > /data/logs/tomcat/data-consumer/spring.log &

但是执行发现，consumer的利用率并不高，如下图：

查了下资料，consumer utilisation 低的原因有三点

1、消费者太少；

2、消费端的ack太慢；

3、消费者太多。

因为我设置了 factory.setAcknowledgeMode(AcknowledgeMode.NONE); 那么就不存在第二种原因，那么我就调整了一下vm option参数，加大速度，很快consumer utilisation一直持续在96%以上，程序运行不到3个小时，数据都已经迁移完毕；

优化后的查询速度如下图:

感谢各位的阅读，以上就是“如何解决mongodb深分页的问题”的内容了，经过本文的学习后，相信大家对如何解决mongodb深分页的问题这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是亿速云，小编将为大家推送更多相关知识点的文章，欢迎关注！