前言 传统关系数据库中都提供了基于row number的分页功能，切换MongoDB后，想要实现分页，则需要修改一下思路。
传统分页思路 假设一页大小为10条。则
1 2 3 4 5 6 7 8 9 10 11 12 //page 1 1-10 //page 2 11-20 //page 3 21-30 ... //page n 10*(n-1) +1 - 10*n
MongoDB提供了skip()和limit()方法。
skip : 跳过指定数量的数据. 可以用来跳过当前页之前的数据，即跳过pageSize*(n-1)。limit : 指定从MongoDB中读取的记录条数，可以当做页面大小pageSize。
所以，分页可以这样做：
1 2 3 4 5 6 7 //Page 1 db.users.find().limit (10) //Page 2 db.users.find().skip(10).limit(10) //Page 3 db.users.find().skip(20).limit(10) ........
问题
看起来，分页已经实现了，但是官方文档并不推荐，说会扫描全部文档，然后再返回结果。
The cursor.skip() method requires the server to scan from the beginning of the input results set before beginning to return results. As the offset increases, cursor.skip() will become slower.
所以，需要一种更快的方式。其实和mysql数量大之后不推荐用limit m,n一样，解决方案是先查出当前页的第一条，然后顺序数pageSize条。MongoDB官方也是这样推荐的。
正确的分页办法 我们假设基于_id的条件进行查询比较。事实上，这个比较的基准字段可以是任何你想要的有序的字段，比如时间戳。
1 2 3 4 5 6 7 8 9 10 11 //Page 1 db.users.find().limit(pageSize); //Find the id of the last document in this page last_id = ... //Page 2 users = db.users.find({ '_id' :{ "$gt" :ObjectId("5b16c194666cd10add402c87")} }).limit(10) //Update the last id with the id of the last document in this page last_id = ...
显然，第一页和后面的不同。对于构建分页API, 我们可以要求用户必须传递pageSize, lastId。
pageSize 页面大小 lastId 上一页的最后一条记录的id，如果不传，则将强制为第一页 降序 _id降序，第一页是最大的，下一页的id比上一页的最后的id还小。
1 2 3 4 5 6 7 8 9 10 11 12 function printStudents (startValue, nPerPage ) let endValue = null ; db.students.find( { _id : { $lt : startValue } } ) .sort( { _id : -1 } ) .limit( nPerPage ) .forEach( student => print( student.name ); endValue = student._id; } ); return endValue; }
升序 _id升序， 下一页的id比上一页的最后一条记录id还大。
1 2 3 4 5 6 7 8 9 10 11 12 function printStudents (startValue, nPerPage ) let endValue = null ; db.students.find( { _id : { $gt : startValue } } ) .sort( { _id : 1 } ) .limit( nPerPage ) .forEach( student => print( student.name ); endValue = student._id; } ); return endValue; }
一共多少条 还有一共多少条和多少页的问题。所以，需要先查一共多少条count.
1 db.users.find().count();
ObjectId的有序性问题 先看ObjectId生成规则：
比如"_id" : ObjectId("5b1886f8965c44c78540a4fc")
取id的前4个字节。由于id是16进制的string，4个字节就是32位，对应id前8个字符。即5b1886f8, 转换成10进制为1528334072. 加上1970，就是当前时间。
事实上，更简单的办法是查看org.mongodb:bson:3.4.3里的ObjectId对象。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 public ObjectId (Date date) this (dateToTimestampSeconds(date), MACHINE_IDENTIFIER, PROCESS_IDENTIFIER, NEXT_COUNTER.getAndIncrement(), false ); } private static int dateToTimestampSeconds (Date time) return (int )(time.getTime() / 1000L ); } public long getTime () return getTimeImpl(); }
MongoDB的ObjectId应该是随着时间而增加的，即后插入的id会比之前的大。但考量id的生成规则，最小时间排序区分是秒，同一秒内的排序无法保证。当然，如果是同一台机器的同一个进程生成的对象，是有序的。
如果是分布式机器，不同机器时钟同步和偏移的问题。所以，如果你有个字段可以保证是有序的，那么用这个字段来排序是最好的。_id则是最后的备选方案。
如果我一定要跳页 上面的分页看起来看理想，虽然确实是，但有个刚需不曾指明—我怎么跳页。
我们的分页数据要和排序键关联，所以必须有一个排序基准来截断记录。而跳页，我只知道第几页，条件不足，无法分页了。
现实业务需求确实提出了跳页的需求，虽然几乎不会有人用，人们更关心的是开头和结尾，而结尾可以通过逆排序的方案转成开头。所以，真正分页的需求应当是不存在的。如果你是为了查找某个记录，那么查询条件搜索是最快的方案。如果你不知道查询条件，通过肉眼去一一查看，那么下一页足矣。
说了这么多，就是想扭转传统分页的概念，在互联网发展的今天，大部分数据的体量都是庞大的，跳页的需求将消耗更多的内存和cpu，对应的就是查询慢。
当然，如果数量不大，如果不介意慢一点，那么skip也不是啥问题，关键要看业务场景。
我今天接到的需求就是要跳页，而且数量很小，那么skip吧，不费事，还快。
来看看大厂们怎么做的
Google最常用了，看起来是有跳页选择的啊。再仔细看，只有10页，多的就必须下一页，并没有提供一共多少页，跳到任意页的选择。这不就是我们的find-condition-then-limit方案吗，只是他的一页数量比较多，前端或者后端把这一页给切成了10份。
同样，看Facebook，虽然提供了总count，但也只能下一页。
其他场景，比如Twitter，微博，朋友圈等，根本没有跳页的概念的。
排序和性能 前面关注于分页的实现原理，但忽略了排序。既然分页，肯定是按照某个顺序进行分页的，所以必须要有排序的。
MongoDB的sort和find组合
1 2 db.bios.find().sort( { name: 1 } ).limit( 5 ) db.bios.find().limit( 5 ).sort( { name: 1 } )
这两个都是等价的，顺序不影响执行顺序。即，都是先find查询符合条件的结果，然后在结果集中排序。
我们条件查询有时候也会按照某字段排序的，比如按照时间排序。查询一组时间序列的数据，我们想要按照时间先后顺序来显示内容，则必须先按照时间字段排序，然后再按照id升序。
1 db.users.find({name: "Ryan"}).sort( { birth: 1, _id: 1 } ).limit( 5 )
我们先按照birth升序，然后birth相同的record再按照_id升序，如此可以实现我们的分页功能了。
多字段排序 1 db.records.sort({ a:1, b:-1})
表示先按照a升序，再按照b降序。即，按照字段a升序，对于a相同的记录，再用b降序，而不是按a排完之后再全部按b排。
示例：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 db.user.find(); 结果： { "_id" : ObjectId("5b1886ac965c44c78540a4fb"), "name" : "a", "age" : 1.0, "id" : "1" } { "_id" : ObjectId("5b1886f8965c44c78540a4fc"), "name" : "a", "age" : 2.0, "id" : "2" } { "_id" : ObjectId("5b1886fa965c44c78540a4fd"), "name" : "b", "age" : 1.0, "id" : "3" } { "_id" : ObjectId("5b1886fd965c44c78540a4fe"), "name" : "b", "age" : 2.0, "id" : "4" } { "_id" : ObjectId("5b1886ff965c44c78540a4ff"), "name" : "c", "age" : 10.0, "id" : "5" }
按照名称升序，然后按照age降序
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 db.user.find({}).sort({name: 1, age: -1}) 结果： { "_id" : ObjectId("5b1886f8965c44c78540a4fc"), "name" : "a", "age" : 2.0, "id" : "2" } { "_id" : ObjectId("5b1886ac965c44c78540a4fb"), "name" : "a", "age" : 1.0, "id" : "1" } { "_id" : ObjectId("5b1886fd965c44c78540a4fe"), "name" : "b", "age" : 2.0, "id" : "4" } { "_id" : ObjectId("5b1886fa965c44c78540a4fd"), "name" : "b", "age" : 1.0, "id" : "3" } { "_id" : ObjectId("5b1886ff965c44c78540a4ff"), "name" : "c", "age" : 10.0, "id" : "5" }
用索引优化排序 到这里必须考虑下性能。
$sort and Memory Restrictions allowDiskUse option to true to enable $sort operations to write to temporary files. See the allowDiskUse option in db.collection.aggregate() method and the aggregate command for details.
Changed in version 2.6: The memory limit for $sort changed from 10 percent of RAM to 100 megabytes of RAM.
从2.6开始，sort只排序100M以内的数据，超过将会报错。可以通过设置allowDiskUse来允许排序大容量数据。
有索引的排序会比没有索引的排序快，所以官方推荐为需要排序的key建立索引。
索引 对于单key排序，建立单独索引
1 db.records.createIndex( { a: 1 } )
索引可以支持同排序和逆序的sort
索引又分升序(1)和降序(-1)，索引定义的排序方向以及逆转方向可以支持sort。对于上述单key索引a，可以支持sort({a:1})升序和sort({a:-1})降序。
对于多字段排序
如果想要使用索引。则可以建立复合(compound index)索引为
1 db.records.createIndex( { a: 1, b:-1 } )
复合索引的字段顺序必须和sort一致
复合多字段索引的顺序要和sort的字段一致才可以走索引。比如索引{a:1, b:1}, 可以支持sort({a:1, b:1})和逆序sort({a:-1, b:-1})， 但是，不支持a，b颠倒。即，不支持sort({b:1, a:1}).
复合索引支持sort同排序和逆序
索引{a:1, b:-1} 可以支持sort({a:1, b:-1}), 也可以支持sort({a:-1, b:1})
复合索引可以前缀子集支持sort
对于多字段复合索引，可以拆分成多个前缀子集。比如{a:1, b:1, c:1}相当于
1 2 3 { a: 1 } { a: 1, b: 1 } { a: 1, b: 1, c: 1 }
示例：
Example Index Prefix db.data.find().sort( { a: 1 } ) { a: 1 } db.data.find().sort( { a: -1 } ) { a: 1 } db.data.find().sort( { a: 1, b: 1 } ) { a: 1, b: 1 } db.data.find().sort( { a: -1, b: -1 } ) { a: 1, b: 1 } db.data.find().sort( { a: 1, b: 1, c: 1 } ) { a: 1, b: 1, c: 1 } db.data.find( { a: { $gt: 4 } } ).sort( { a: 1, b: 1 } ) { a: 1, b: 1 }
复合索引的非前缀子集可以支持sort，前提是前缀子集的元素要在find的查询条件里是equals
这个条件比较绕口，复合索引的非前缀子集，只要find和sort的字段要组成索引前缀，并且find里的条件必须是相等。
示例
Example Index Prefix db.data.find( { a: 5 } ).sort( { b: 1, c: 1 } ) { a: 1 , b: 1, c: 1 } db.data.find( { b: 3, a: 4 } ).sort( { c: 1 } ) { a: 1, b: 1, c: 1 } db.data.find( { a: 5, b: { $lt: 3} } ).sort( { b: 1 } ) { a: 1, b: 1 }
find和sort的字段加起来满足前缀子集，find条件中可以使用其他字段进行非equals比较。
对于既不是前缀子集，也不是find相等条件的。索引无效。比如，对于索引{a:1, b:1, c:1}。以下两种方式不走索引。
1 2 db.data.find( { a: { $gt: 2 } } ).sort( { c: 1 } ) db.data.find( { c: 5 } ).sort( { c: 1 } )
Java代码分页 由于确实有跳页的需求，目前还没有发现性能问题，仍旧采用skip做分页，当然也兼容条件分页
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 public PageResult<StatByClientRs> findByDurationPage (FindByDurationPageRq rq) final Criteria criteriaDefinition = Criteria.where("duration" ).is(rq.getDuration()); final Query query = new Query( criteriaDefinition ).with(new Sort(Lists.newArrayList(new Order(Direction.ASC, "_id" )))); long total = mongoTemplate.count(query, StatByClient.class); Integer pageSize = rq.getPageSize(); Integer pageNum = rq.getPageNum(); String lastId = rq.getLastId(); final Integer pages = (int ) Math.ceil(total / (double ) pageSize); if (pageNum<=0 || pageNum> pages) { pageNum = 1 ; } List<StatByClient> statByClientList; if (StringUtils.isNotBlank(lastId)) { if (pageNum != 1 ) { criteriaDefinition.and("_id" ).gt(new ObjectId(lastId)); } query.limit(pageSize); statByClientList = mongoTemplate .find(query, StatByClient.class); } else { int skip = pageSize * (pageNum - 1 ); query.skip(skip).limit(pageSize); statByClientList = mongoTemplate.find(query, StatByClient.class); } PageResult<StatByClientRs> pageResult = new PageResult<>(); pageResult.setTotal(total); pageResult.setPages(pages); pageResult.setPageSize(pageSize); pageResult.setPageNum(pageNum); pageResult.setList(mapper.mapToListRs(statByClientList)); return pageResult; }
这个示例中，目标是根据duration查询list，结果集进行分页。当请求体中包含lastId，那就走下一页方案。如果想要跳页，就不传lastId，随便你跳吧。
抽取分页代码为公共工具类 考虑分页需求的旺盛，每个集合都这样写感觉比较麻烦，而且容易出错。我们来把这个封装成单独一个PageHelper
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 import com.google.common.collect.Lists;import com.shuwei.d2.message.PageResult;import java.util.List;import java.util.function.Function;import java.util.stream.Collectors;import org.apache.commons.lang3.StringUtils;import org.bson.types.ObjectId;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.domain.Sort;import org.springframework.data.domain.Sort.Direction;import org.springframework.data.domain.Sort.Order;import org.springframework.data.mongodb.core.MongoTemplate;import org.springframework.data.mongodb.core.query.Criteria;import org.springframework.data.mongodb.core.query.Query;import org.springframework.stereotype.Component;@Component public class MongoPageHelper public static final int FIRST_PAGE_NUM = 1 ; public static final String ID = "_id" ; private final MongoTemplate mongoTemplate; @Autowired public MongoPageHelper (MongoTemplate mongoTemplate) this .mongoTemplate = mongoTemplate; } public <T> PageResult<T> pageQuery (Query query, Class<T> entityClass, Integer pageSize, Integer pageNum) return pageQuery(query, entityClass, Function.identity(), pageSize, pageNum, null ); } public <T, R> PageResult<R> pageQuery (Query query, Class<T> entityClass, Function<T, R> mapper, Integer pageSize, Integer pageNum) return pageQuery(query, entityClass, mapper, pageSize, pageNum, null ); } public <T, R> PageResult<R> pageQuery (Query query, Class<T> entityClass, Function<T, R> mapper, Integer pageSize, Integer pageNum, String lastId) long total = mongoTemplate.count(query, entityClass); final Integer pages = (int ) Math.ceil(total / (double ) pageSize); if (pageNum <= 0 || pageNum > pages) { pageNum = FIRST_PAGE_NUM; } final Criteria criteria = new Criteria(); if (StringUtils.isNotBlank(lastId)) { if (pageNum != FIRST_PAGE_NUM) { criteria.and(ID).gt(new ObjectId(lastId)); } query.limit(pageSize); } else { int skip = pageSize * (pageNum - 1 ); query.skip(skip).limit(pageSize); } final List<T> entityList = mongoTemplate .find(query.addCriteria(criteria) .with(new Sort(Lists.newArrayList(new Order(Direction.ASC, ID)))), entityClass); final PageResult<R> pageResult = new PageResult<>(); pageResult.setTotal(total); pageResult.setPages(pages); pageResult.setPageSize(pageSize); pageResult.setPageNum(pageNum); pageResult.setList(entityList.stream().map(mapper).collect(Collectors.toList())); return pageResult; } }
对了，还有PageResult对象
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 import com.fasterxml.jackson.annotation.JsonInclude;import com.fasterxml.jackson.annotation.JsonInclude.Include;import io.swagger.annotations.ApiModelProperty;import java.util.List;import lombok.AllArgsConstructor;import lombok.Data;import lombok.NoArgsConstructor;@Data @AllArgsConstructor @NoArgsConstructor @JsonInclude(Include.NON_NULL) public class PageResult <T > @ApiModelProperty("页码，从1开始") private Integer pageNum; @ApiModelProperty("页面大小") private Integer pageSize; @ApiModelProperty("总数") private Long total; @ApiModelProperty("总页数") private Integer pages; @ApiModelProperty("数据") private List<T> list; }
使用工具类 最初的查询语句，业务逻辑和分页逻辑分开。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 public PageResult<StatByClientRs> findByDurationPage (FindByDurationPageRq rq) final Criteria criteriaDefinition = Criteria.where("duration" ).is(rq.getDuration()); if (rq.getClientId() == null ) { criteriaDefinition.and("clientId" ).ne(AllFlag.ALL_CLIENT); } else { criteriaDefinition.and("clientId" ).is(rq.getClientId()); } final Query query = new Query(criteriaDefinition) .with(new Sort(Lists.newArrayList(new Order(Direction.ASC, "_id" )))); return mongoPageHelper.pageQuery(query, StatByClient.class, mapper::mapToRs, rq.getPageSize(), rq.getPageNum(), rq.getLastId()); }
把工具类共享到maven仓库 新建一个maven项目，https://github.com/Ryan-Miao/mongo-page-helper
修改并提取刚才的工具类。
如何使用 必须结合spring-boot-starter-data-mongodb来使用.
在pom里添加repository
1 2 3 4 5 6 <repositories > <repository > <id > jitpack.io</id > <url > https://jitpack.io</url > </repository > </repositories >
再引入依赖
1 2 3 4 5 <dependency > <groupId > com.github.Ryan-Miao</groupId > <artifactId > mongo-page-helper</artifactId > <version > 1.0</version > </dependency >
配置Configuration
1 2 3 4 5 6 7 8 9 10 11 @Configuration public class MongoConfiguration @Autowired private MongoTemplate mongoTemplate; @Bean public MongoPageHelper mongoPageHelper () return new MongoPageHelper(mongoTemplate); } }
然后就可以使用MongoPageHelper来注入了。
参考 
