榆社网站建设见网页制作的公司网站

本文主要介绍limit 分页的弊端及线上应该怎么用

LIMIT M,N

平时经常见到使用 <limit m,n>+ 合适的 order by 来实现分页查询，这样做到底性能如何呢？

先来简单分析下，然后再实际验证一下。

1. 无索引条件下，需要做大量的文件排序操作，性能将会非常糟糕；
2. 有索引条件下，刚开始的分页查询效率会比较理想，但越往后，分页查询的性能就越差。

这主要是因为，在使用 LIMIT 的时候，偏移量 M 在分页越靠后的时候，值就越大，数据库检索的数据也就越多。
例如 LIMIT 90000,10 这样的查询，数据库需要查询 90010 条记录，最后返回 10 条记录。也就是说将会有 90000 条记录被查询出来没有被使用到。

下面我们来验证下
首先创建一张会员表，表结构如下

CREATE TABLE `member` (`id` int(11) NOT NULL AUTO_INCREMENT,`member_name` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL,`member_phone` varchar(20) COLLATE utf8mb4_unicode_ci DEFAULT NULL,`join_date` datetime DEFAULT CURRENT_TIMESTAMP,`member_id` bigint(20) NOT NULL,PRIMARY KEY (`id`),UNIQUE KEY `idx_member_id` (`member_id`)
)

插入 10 万条数据

DELIMITER //CREATE PROCEDURE InsertMember()
BEGINDECLARE i INT DEFAULT 0;WHILE i < 100000 DO-- 为member_id生成一个10位随机数SET @random_member_id = FLOOR(RAND() * 9000000000 + 1000000000) + i*RAND();-- 插入数据INSERT INTO member (member_name, member_phone, member_id)VALUES (CONCAT('Member', LPAD(i + 1, 5, '0')), -- 会员姓名，编号后面跟5个0CONCAT('13', LPAD(RAND()*(9999999999-1000000000+1)+1000000000, 10, '0')), -- 随机生成电话号码@random_member_id -- 随机生成的会员编号);-- 增加循环计数器SET i = i + 1;END WHILE;
END //DELIMITER ;

执行存储过程

CALL InsertMember();

验证 limit 查询

执行sql

select * from member order by member_id limit 90000, 10;

可以看到，所用查询时间为 0.227s，相对来说时间偏长了。

子查询优化

先查询出所需要的 10 行数据中的最小 ID 值，然后通过偏移量返回所需要的 10 行数据，可以通过索引覆盖扫描，使用子查询的方式来实现分页查询

SELECT *
FROMmember
WHEREid > (SELECT idFROMmemberORDER BY member_idLIMIT 90000 , 1)
LIMIT 10;

执行时间 0.024s

线上分页

那么在实际的生产环境中，该怎么使用呢？下面我来介绍下我当时是怎么做的。
核心思想就是：分段查询

假如有个订单表，在 【2024-01-01 00:00:00,2024-01-02 00:00:00】有12万条数据， 前 11 个小时段有接近于 1 万条数据，第 12 个小时段有大于 1 万条数据。

现在我们采用分时间段查询，间隔为 1 小时，每次查询 2000 条，那么每个小时段需要查询 5-6次。

先贴出 SQL 代码，方便查看

<select id="grabBizDataSlice" resultMap="BaseResultMap">select<include refid="Base_Column_List"/>from orderwhere update_time &gt;= \#{startTime} and update_time &lt; \#{endTime}and status = 'PROCESS'and id > \#{startRow}order by idlimit  \#{pageSize}</select>

第一个小时

第一次查询

时间段：【2024-01-01 00:00:00，2024-01-01 01:00:00】，
startRow：0
pageSize：2000

第二次查询

时间段：【2024-01-01 00:00:00，2024-01-01 01:00:00】，
startRow：2000
pageSize：2000

第三次查询

时间段：【2024-01-01 00:00:00，2024-01-01 01:00:00】，
startRow： 4000
pageSize：2000

第四次查询

时间段：【2024-01-01 00:00:00，2024-01-01 01:00:00】，
startRow：6000
pageSize：2000

第五次查询

时间段：【2024-01-01 00:00:00，2024-01-01 01:00:00】，
startRow：> 8000
pageSize：2000

注意：第 5 次查询的时候，实际返回的数据量总量已经小于 2000 条了，此时我们就可以判断到第一个小时段的数据已经查询结束了,然后开始第二个时间段的查询，道理是一样的。

redis 存储分段条件

通过上面可以看出来，我们需要有一个地方来保存每次查询的条件的。
这里我是采用的 redis hash 结构。

private final  Map<String, String> bizIdxKeyMap = new HashMap<>();
private final Integer pageSize = 2000;List<B> bizDataList = ...; //从数据库查询的记录
Long pageIdx = bizDataList.size() == pageSize ? bizDataList.get(pageSize - 1).getId() : -1;
bizIdxKeyMap.put("sliceStartCache", sliceStartTime);
bizIdxKeyMap.put("sliceEndCache", sliceEndTime);
bizIdxKeyMap.put("pageIdxCache", pageIdx.toString());
redisCluster.hmset(bizIdxKey, bizIdxKeyMap);

从这里可以看到，当pageIdx = -1时，代表本时间段查询结束了。在下次循环时，再从 redis 中取出来这三个字段 sliceStartCache、sliceEndCache、pageIdxCache。

完整代码

class InitController {@Autowiredprivate BizCommonService bizCommonService;@Autowiredprivate OrderInitServiceImpl orderInitServiceImpl;void calculationFlow(Date startTime, Date endTime) {bizCommonService.initFinanceCalculationCycle(startTime, endTime);orderInitServiceImpl.orderInit();}
}@Service
public class OrderInitServiceImpl{@Autowiredprivate OdsPackOrderDAO odsPackOrderDAO;@Autowiredprivate BizCommonService  bizCommonService;public void orderInit() throws InterruptedException {while(true){String packOrderCalculationSwitch = redisCluster.get("pack_order_switch");if(packOrderCalculationSwitch != null && packOrderCalculationSwitch.equals("switch_off")){break; //查询结束}List<OdsPackOrderDO> odsPackOrderDOList = bizCommonService.grabBizDataSlice(3,TimeUnit.MINUTES, 2000, odsPackOrderDAO, null);// 对查询出来的odsPackOrderDOList做一些业务逻辑}}}@Component
public class BizCommonServicelImpl{@Autowiredprotected RedisCluster redisCluster;private Date financeCycleStartTime;private Date financeCycleEndTime;private final  Map<String, String> bizIdxKeyMap = new HashMap<>();private final static Calendar calendar= Calendar.getInstance();public void initFinanceCalculationCycle(Date startTime, Date endTime) {this.financeCycleStartTime = startTime;this.financeCycleEndTime = endTime;}public List<B> grabBizData(@NonNull Integer interval, TimeUnit intervalUnit, @NonNull Integer pageSize, BD bizDataSource, @Nullable Object customParam){try{String bizIdxKey = "order_index_key"; // 分页条件键String bizSwitchKey = "pack_order_switch"; // 查询终止状态键// 从 redis 查询分页条件键List<String> bizIdxCache = redisCluster.hmget(bizIdxKey, "sliceStartCache", "sliceEndCache", "pageIdxCache");Long pageIdx;Date sliceEndTime;Date sliceStartTime;if(bizIdxCache.get(2) == null ||  bizIdxCache.get(2).equals("-1")){pageIdx = 0L;if(bizIdxCache.get(0) == null){sliceStartTime = financeCycleStartTime;sliceEndTime = timer(sliceStartTime, interval, intervalUnit);}else{sliceStartTime = DateUtils.getDateByMySQLDateTimeString(bizIdxCache.get(1));sliceEndTime = timer(sliceStartTime, interval, intervalUnit);}}else{sliceStartTime = DateUtils.getDateByMySQLDateTimeString(bizIdxCache.get(0));sliceEndTime = DateUtils.getDateByMySQLDateTimeString(bizIdxCache.get(1));pageIdx = Long.valueOf(bizIdxCache.get(2));}// 判断结束标志if(sliceStartTime != null && (sliceStartTime.after(financeCycleEndTime) || sliceStartTime.equals(financeCycleEndTime))){redisCluster.set("pack_order_switch", SWITCH_OFF);return null;}List<B> bizDataList;if(customParam == null) {bizDataList = bizDataSource.grabBizDataSlice(sliceStartTime,sliceEndTime.after(financeCycleEndTime) ? financeCycleEndTime : sliceEndTime,pageIdx,pageSize);}else{bizDataList = bizDataSource.grabBizDataSliceByCustomParam(sliceStartTime,sliceEndTime.after(financeCycleEndTime) ? financeCycleEndTime : sliceEndTime,pageIdx,pageSize,customParam);}pageIdx = bizDataList.size() == pageSize ? bizDataList.get(pageSize - 1).getId() : -1;bizIdxKeyMap.put("sliceStartCache", sliceStartTime);bizIdxKeyMap.put("sliceEndCache", sliceEndTime);bizIdxKeyMap.put("pageIdxCache", pageIdx.toString());redisCluster.hmset("order_index_key", bizIdxKeyMap);return bizDataList;}catch (Exception e){return null;}}private Date timer(Date currentTime, Integer interval, TimeUnit intervalUnit){calendar.setTime(currentTime);if(intervalUnit == TimeUnit.DAYS){calendar.add(Calendar.DATE, interval);}else if(intervalUnit == TimeUnit.HOURS){calendar.add(Calendar.HOUR, interval);}else if(intervalUnit == TimeUnit.MINUTES){calendar.add(Calendar.MINUTE, interval);}else if(intervalUnit == TimeUnit.SECONDS){calendar.add(Calendar.SECOND, interval);}else {throw new RuntimeException("");}return calendar.getTime();}
}

总结

采取合理的分页方式可以有效的提升系统性能，应根据实际情况选择适合自己的方式。
欢迎各位老师分享工作中是怎么使用的，可以交流交流。