springboot分页查询并行优化实践

　　　　　　　　　　　　——基于异步优化与 MyBatis-Plus 分页插件思想的实践
适用场景

• 数据量较大的单表分页查询
  
• 较复杂的多表关联查询，包含group by等无法举行count优化较耗时的分页查询
  

技术栈

• 核心框架：Spring Boot + MyBatis-Plus
  
• 异步编程：JDK 8+ 的 CompletableFuture 
  
• 数据库：MySQL 8.0
  
• 线程池：自定义线程池管理并行任务（如 ThreadPoolTaskExecutor）
  

实现思绪

解决传统分页查询中 串行实行 COUNT 与数据查询 的性能瓶颈，通过 并行化 淘汰总耗时，同时兼容复杂查询场景（如多表关联、DISTINCT 等）
兼容mybatisPlus分页参数，复用 IPage 接口定义分页参数（当前页、每页条数），
借鉴 MyBatis-Plus 的 PaginationInnerInterceptor，通过实现 MyBatis 的 Interceptor 接口，拦截 Executor#query 方法，动态修改 SQL，
sql优化适配：COUNT 优化：自动移除 ORDER BY，保存 GROUP BY 和 DISTINCT（需包裹子查询），数据查询：保存完整 SQL 逻辑，仅追加 LIMIT 和 OFFSET。
直接上代码

使用简单
调用查询方法前赋值page对象属性total大于0数值则可进入自定义分页查询方案。

1. //示例代码
2. Page<User> page = new Page<>(1,10);
3. page.setTotal(1L);

复制代码

线程池配置

1. @Configuration
2. public class ThreadPoolTaskExecutorConfig {
4.     public static final Integer CORE_POOL_SIZE = 20;
5.     public static final Integer MAX_POOL_SIZE = 40;
6.     public static final Integer QUEUE_CAPACITY = 200;
7.     public static final Integer KEEP_ALIVE_SECONDS = 60;
9.     @Bean("threadPoolTaskExecutor")
10.     public ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
11.         ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
12.         //核心线程数
13.         threadPoolTaskExecutor.setCorePoolSize(CORE_POOL_SIZE);
14.         //线程池最大线程数
15.         threadPoolTaskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
16.         //队列容量
17.         threadPoolTaskExecutor.setQueueCapacity(QUEUE_CAPACITY);
18.         //线程空闲存活时间
19.         threadPoolTaskExecutor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
20.         //线程前缀
21.         threadPoolTaskExecutor.setThreadNamePrefix("commonTask-");
22.         //拒绝策略
23.         threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
24.         //线程池初始化
25.         threadPoolTaskExecutor.initialize();
27.         return threadPoolTaskExecutor;
28.     }
30.     @Bean("countAsyncThreadPool")
31.     public ThreadPoolTaskExecutor getCountAsyncThreadPool() {
32.         ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
33.         //核心线程数，根据负载动态调整
34.         threadPoolTaskExecutor.setCorePoolSize(6);
35.         //线程池最大线程数，根据负载动态调整
36.         threadPoolTaskExecutor.setMaxPoolSize(12);
37.         //队列容量  队列容量不宜过多，根据负载动态调整
38.         threadPoolTaskExecutor.setQueueCapacity(2);
39.         //线程空闲存活时间
40.         threadPoolTaskExecutor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
41.         //线程前缀
42.         threadPoolTaskExecutor.setThreadNamePrefix("countAsync-");
43.         //拒绝策略  队列满时由调用者主线程执行
44.         threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
45.         //线程池初始化
46.         threadPoolTaskExecutor.initialize();
48.         return threadPoolTaskExecutor;
49.     }
50. }

复制代码

mybatis-plus配置类

1. @Configuration
2. @MapperScan("com.xxx.mapper")
3. public class MybatisPlusConfig {
5.     @Resource
6.     ThreadPoolTaskExecutor countAsyncThreadPool;
7.     @Resource
8.     ApplicationContext applicationContext;
10.     @Bean
11.     public MybatisPlusInterceptor mybatisPlusInterceptor() {
12.         MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
13.         interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL));
14.         return interceptor;
15.     }
18.     @Bean
19.     public PageParallelQueryInterceptor pageParallelQueryInterceptor() {
20.         PageParallelQueryInterceptor pageParallelQueryInterceptor = new PageParallelQueryInterceptor();
21.         pageParallelQueryInterceptor.setCountAsyncThreadPool(countAsyncThreadPool);
22.         pageParallelQueryInterceptor.setApplicationContext(applicationContext);
23.         return pageParallelQueryInterceptor;
24.     }
25. }

复制代码

自定义mybatis拦截器

1. package com.example.dlock_demo.interceptor;
3. import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
4. import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
5. import lombok.extern.slf4j.Slf4j;
6. import net.sf.jsqlparser.JSQLParserException;
7. import net.sf.jsqlparser.expression.Expression;
8. import net.sf.jsqlparser.parser.CCJSqlParserUtil;
9. import net.sf.jsqlparser.statement.select.*;
10. import org.apache.ibatis.builder.StaticSqlSource;
11. import org.apache.ibatis.cache.CacheKey;
12. import org.apache.ibatis.executor.Executor;
13. import org.apache.ibatis.mapping.BoundSql;
14. import org.apache.ibatis.mapping.MappedStatement;
15. import org.apache.ibatis.mapping.ResultMap;
16. import org.apache.ibatis.plugin.Interceptor;
17. import org.apache.ibatis.plugin.Intercepts;
18. import org.apache.ibatis.plugin.Invocation;
19. import org.apache.ibatis.plugin.Signature;
20. import org.apache.ibatis.session.*;
21. import org.springframework.context.ApplicationContext;
22. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
24. import java.lang.reflect.Field;
25. import java.lang.reflect.Method;
26. import java.sql.SQLException;
27. import java.util.*;
28. import java.util.concurrent.CompletableFuture;
29. import java.util.concurrent.CompletionException;
30. import java.util.concurrent.ConcurrentHashMap;
33. /**
34. * Mybatis-分页并行查询拦截器
35. *
36. * @author shf
37. */
38. @Intercepts({
39.         @Signature(type = Executor.class, method = "query",
40.                 args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class}),
41.         @Signature(type = Executor.class, method = "query",
42.                 args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class, CacheKey.class, BoundSql.class})
43. })
44. @Slf4j
45. public class PageParallelQueryInterceptor implements Interceptor {
46.     /**
47.      * 用于数据库并行查询线程池
48.      */
49.     private ThreadPoolTaskExecutor countAsyncThreadPool;
50.     /**
51.      * 容器上下文
52.      */
53.     private ApplicationContext applicationContext;
55.     private static final String LONG_RESULT_MAP_ID = "twoPhase-Long-ResultMap";
56.     private static final Map<String, MappedStatement> twoPhaseMsCache = new ConcurrentHashMap();
58.     public void setCountAsyncThreadPool(ThreadPoolTaskExecutor countAsyncThreadPool) {
59.         this.countAsyncThreadPool = countAsyncThreadPool;
60.     }
62.     public void setApplicationContext(ApplicationContext applicationContext) {
63.         this.applicationContext = applicationContext;
64.     }
66.     @Override
67.     public Object intercept(Invocation invocation) throws Throwable {
68.         Object[] args = invocation.getArgs();
69.         MappedStatement ms = (MappedStatement) args[0];
70.         Object parameter = args[1];
72.         //获取分页参数
73.         Page<?> page = getPageParameter(parameter);
74.         if (page == null || page.getSize() <= 0 || !page.searchCount() || page.getTotal() == 0) {
75.             return invocation.proceed();
76.         }
77.         //获取Mapper方法(注解形式 需利用反射且只能应用在mapper接口层，不推荐使用)
78.         /*Method method = getMapperMethod(ms);
79.         if (method == null || !method.isAnnotationPresent(PageParallelQuery.class)) {
80.             return invocation.proceed();
81.         }*/
83.         BoundSql boundSql = ms.getBoundSql(parameter);
84.         String originalSql = boundSql.getSql();
85.         //禁用mybatis plus PaginationInnerInterceptor count查询
86.         page.setSearchCount(false);
87.         page.setTotal(0);
88.         args[2] = RowBounds.DEFAULT;
89.         CompletableFuture<Long> countFuture = resolveCountCompletableFuture(invocation, originalSql);
90.         //limit查询
91.         long startTime = System.currentTimeMillis();
92.         Object proceed = invocation.proceed();
93.         log.info("原SQL数据查询-耗时={}", System.currentTimeMillis() - startTime);
94.         page.setTotal(countFuture.get());
96.         return proceed;
97.     }
99.     private CompletableFuture<Long> resolveCountCompletableFuture(Invocation invocation, String originalSql) {
100.         return CompletableFuture.supplyAsync(() -> {
101.             try {
102.                 //查询总条数
103.                 long startTime = System.currentTimeMillis();
104.                 long total = executeCountQuery(originalSql, invocation);
105.                 log.info("分页并行查询COUNT总条数[{}]-耗时={}", total, System.currentTimeMillis() - startTime);
106.                 return total;
107.             } catch (Throwable e) {
108.                 log.error("page parallel query exception:", e);
109.                 throw new CompletionException(e);
110.             }
111.         }, countAsyncThreadPool).exceptionally(throwable -> {
112.             log.error("page parallel query exception:", throwable);
113.             return 0L;
114.         });
115.     }
117.     private CompletableFuture<Object> resolveOriginalProceedCompletableFuture(Invocation invocation) {
118.         return CompletableFuture.supplyAsync(() -> {
119.             try {
120.                 long startTime = System.currentTimeMillis();
121.                 Object proceed = invocation.proceed();
122.                 log.info("原SQL数据查询-耗时={}", System.currentTimeMillis() - startTime);
123.                 return proceed;
124.             } catch (Throwable e) {
125.                 throw new CompletionException(e);
126.             }
127.         }, countAsyncThreadPool).exceptionally(throwable -> {
128.             log.error("page parallel query original proceed exception:", throwable);
129.             return null;
130.         });
131.     }
133.     /**
134.      * 执行count查询
135.      */
136.     private long executeCountQuery(String originalSql, Invocation invocation)
137.             throws JSQLParserException, SQLException {
139.         //解析并修改SQL为count查询
140.         Select countSelect = (Select) CCJSqlParserUtil.parse(originalSql);
141.         PlainSelect plainSelect = (PlainSelect) countSelect.getSelectBody();
143.         //修改select为count(*)
144.         /*plainSelect.setSelectItems(Collections.singletonList(
145.                 new SelectExpressionItem(new Function("COUNT", new Column("*")))
146.         );*/
147.         // 移除排序和分页
148.         Distinct distinct = plainSelect.getDistinct();
149.         GroupByElement groupBy = plainSelect.getGroupBy();
150.         String countSql = "";
151.         if (groupBy == null && distinct == null) {
152.             Expression countFuncExpression = CCJSqlParserUtil.parseExpression("COUNT(*)");
153.             plainSelect.setSelectItems(Collections.singletonList(
154.                     new SelectExpressionItem(countFuncExpression)));
155.             plainSelect.setOrderByElements(null);
156.             countSql = plainSelect.toString();
157.         } else if (groupBy != null) {
158.             plainSelect.setLimit(null);
159.             plainSelect.setOffset(null);
160.             countSql = "SELECT COUNT(*) FROM (" + plainSelect + ") TOTAL";
161.         } else {
162.             plainSelect.setOrderByElements(null);
163.             plainSelect.setLimit(null);
164.             plainSelect.setOffset(null);
165.             countSql = "SELECT COUNT(*) FROM (" + plainSelect + ") TOTAL";
166.         }
167.         //执行count查询
168.         return doCountQuery(invocation, countSql);
169.     }
171.     /**
172.      * 执行修改后的COUNT(*)-SQL查询
173.      */
174.     @SuppressWarnings("unchecked")
175.     private Long doCountQuery(Invocation invocation, String modifiedSql) {
176.         //Executor executor = (Executor) invocation.getTarget();
177.         //创建新会话（自动获取新连接）
178.         Executor executor;
179.         SqlSessionFactory sqlSessionFactory = applicationContext.getBean(SqlSessionFactory.class);
180.         try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.SIMPLE)) {
181.             //com.alibaba.druid.pool.DruidPooledConnection
182.             System.out.println("新会话Connection class: " + sqlSession.getConnection().getClass().getName());
183.             Field executorField = sqlSession.getClass().getDeclaredField("executor");
184.             executorField.setAccessible(true);
185.             executor = (Executor) executorField.get(sqlSession);
187.             Object[] args = invocation.getArgs();
188.             MappedStatement originalMs = (MappedStatement) args[0];
189.             Object parameter = args[1];
190.             //创建新的查询参数
191.             Map<String, Object> newParameter = new HashMap<>();
192.             if (parameter instanceof Map) {
193.                 // 复制原始参数但移除分页参数
194.                 Map<?, ?> originalParams = (Map<?, ?>) parameter;
195.                 originalParams.forEach((k, v) -> {
196.                     if (!(v instanceof Page)) {
197.                         newParameter.put(k.toString(), v);
198.                     }
199.                 });
200.             }
201.             //创建新的BoundSql
202.             BoundSql originalBoundSql = originalMs.getBoundSql(newParameter);
203.             BoundSql newBoundSql = new BoundSql(originalMs.getConfiguration(), modifiedSql, originalBoundSql.getParameterMappings(), newParameter);
204.             //复制原始参数值
205.             originalBoundSql.getParameterMappings().forEach(mapping -> {
206.                 String prop = mapping.getProperty();
207.                 if (mapping.getJavaType().isInstance(newParameter)) {
208.                     newBoundSql.setAdditionalParameter(prop, newParameter);
209.                 } else if (newParameter instanceof Map) {
210.                     Object value = ((Map<?, ?>) newParameter).get(prop);
211.                     newBoundSql.setAdditionalParameter(prop, value);
212.                 }
213.             });
214.             //创建新的BoundSql
215.             /*BoundSql originalBoundSql = originalMs.getBoundSql(parameter);
216.             BoundSql newBoundSql = new BoundSql(originalMs.getConfiguration(), modifiedSql,
217.                     originalBoundSql.getParameterMappings(), parameter);*/
218.             Configuration configuration = originalMs.getConfiguration();
219.             //创建临时ResultMap
220.             ResultMap resultMap = new ResultMap.Builder(
221.                     configuration,
222.                     LONG_RESULT_MAP_ID,
223.                     //强制指定结果类型
224.                     Long.class,
225.                     //自动映射列到简单类型
226.                     Collections.emptyList()
227.             ).build();
228.             if (!configuration.hasResultMap(LONG_RESULT_MAP_ID)) {
229.                 configuration.addResultMap(resultMap);
230.             }
232.             String countMsId = originalMs.getId() + "_countMsId";
233.             MappedStatement mappedStatement = twoPhaseMsCache.computeIfAbsent(countMsId, (key) ->
234.                     this.getNewMappedStatement(modifiedSql, originalMs, newBoundSql, resultMap, countMsId));
235.             //执行查询
236.             List<Object> result = executor.query(mappedStatement, newParameter, RowBounds.DEFAULT, (ResultHandler<?>) args[3]);
237.             long total = 0L;
238.             if (CollectionUtils.isNotEmpty(result)) {
239.                 Object o = result.get(0);
240.                 if (o != null) {
241.                     total = Long.parseLong(o.toString());
242.                 }
243.             }
244.             return total;
245.         } catch (Throwable e) {
246.             log.error("分页并行查询-executeCountQuery异常:", e);
247.         }
248.         return 0L;
249.     }
251.     private MappedStatement getNewMappedStatement(String modifiedSql, MappedStatement originalMs, BoundSql newBoundSql,
252.                                                   ResultMap resultMap, String msId) {
253.         //创建新的MappedStatement
254.         MappedStatement.Builder builder = new MappedStatement.Builder(
255.                 originalMs.getConfiguration(),
256.                 msId,
257.                 new StaticSqlSource(originalMs.getConfiguration(), modifiedSql, newBoundSql.getParameterMappings()),
258.                 originalMs.getSqlCommandType()
259.         );
260.         //复制重要属性
261.         builder.resource(originalMs.getResource())
262.                 .fetchSize(originalMs.getFetchSize())
263.                 .timeout(originalMs.getTimeout())
264.                 .statementType(originalMs.getStatementType())
265.                 .keyGenerator(originalMs.getKeyGenerator())
266.                 .keyProperty(originalMs.getKeyProperties() == null ? null : String.join(",", originalMs.getKeyProperties()))
267.                 .resultMaps(resultMap == null ? originalMs.getResultMaps() : Collections.singletonList(resultMap))
268.                 .parameterMap(originalMs.getParameterMap())
269.                 .resultSetType(originalMs.getResultSetType())
270.                 .cache(originalMs.getCache())
271.                 .flushCacheRequired(originalMs.isFlushCacheRequired())
272.                 .useCache(originalMs.isUseCache());
273.         return builder.build();
274.     }
276.     /**
277.      * 获取分页参数
278.      */
279.     private Page<?> getPageParameter(Object parameter) {
280.         if (parameter instanceof Map) {
281.             Map<?, ?> paramMap = (Map<?, ?>) parameter;
282.             return (Page<?>) paramMap.values().stream()
283.                     .filter(p -> p instanceof Page)
284.                     .findFirst()
285.                     .orElse(null);
286.         }
287.         return parameter instanceof Page ? (Page<?>) parameter : null;
288.     }
290.     /**
291.      * 获取Mapper方法
292.      */
293.     private Method getMapperMethod(MappedStatement ms) {
294.         try {
295.             String methodName = ms.getId().substring(ms.getId().lastIndexOf(".") + 1);
296.             Class<?> mapperClass = Class.forName(ms.getId().substring(0, ms.getId().lastIndexOf(".")));
297.             return Arrays.stream(mapperClass.getMethods())
298.                     .filter(m -> m.getName().equals(methodName))
299.                     .findFirst()
300.                     .orElse(null);
301.         } catch (ClassNotFoundException e) {
302.             return null;
303.         }
304.     }
305. }

复制代码

注意事项

有人可能会担心并行查询，在高并发场景可能会导致count查询与limit数据查询不一致，但其实只要没有锁，只要是分开的两条sql查询，原mybatisplus分页插件也一样面临这个题目。
count优化没有举行join语句判定优化，相当于主动关闭了page.setOptimizeJoinOfCountSql(false);在一对多等场景可能会造成count查询有误，Mybatisplus官网也有干系提示，所以这里干脆舍弃了。
mybatisplus版本差别，可能会导致JsqlParser所使用的api有所差别，需要本身对应版本修改下。本篇版本使用的3.5.1
关于线程池的线程数设置顺便提一下：
网上流行一个说法：
1. CPU 密集型任务
特点：任务主要消耗 CPU 资源（如复杂计算、图像处理）。
线程数建议：

• 核心线程数：CPU 核心数 + 1（或等于CPU核心数，避免上下文切换过多）。
  
• 最大线程数：与核心线程数相同（防止过多线程竞争 CPU）。
  

2. I/O 密集型任务
特点：任务涉及大量等待（如网络哀求、数据库读写）。
线程数建议：

• 核心线程数：2 * CPU 核心数（确保正常负载下的高效处理）。
  
• 最大线程数：根据体系资源调整（用于应对突发高并发）。
  

其实这个说法来源于一个经验公式推导而来：
threads = CPU核心数 * (1 + 平均等待时间 / 平均计算时间)
《Java 虚拟机并发编程》中先容

 
另一篇：《Java Concurrency in Practice》即《java并发编程实践》，给出的线程池巨细的估算公式：
 

Nthreads=Ncpu*Ucpu*(1+w/c)，其中 Ncpu=CPU核心数，Ucpu=cpu使用率，0~1；W/C=等待时间与计算时间的比率
细致推导两个公式，其实类似，在cpu使用率达100%时，其实结论是一致的，这时间计算线程数的公式就成了，Nthreads=Ncpu*100%*(1+w/c) =Ncpu*(1+w/c)。
那么在实践应用中计算的公式就出来了，【以下推算，不思量内存消耗等方面】，如下：
1、针对IO密集型，阻塞耗时w一般都是计算耗时几倍c，假设阻塞耗时=计算耗时的情况下，Nthreads=Ncpu*(1+1)=2Ncpu,所以这种情况下，建议思量2倍的CPU核心数做为线程数
2、对于计算密集型，阻塞耗时趋于0，即w/c趋于0，公式Nthreads = Ncpu。
现实应用时要思量同时设置了几个隔离线程池，另外tomcat自带的线程池也会共享宿主机公共资源。
 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。