<導(dǎo)航

springboot分頁(yè)查詢并行優(yōu)化實(shí)踐

　　　　　　　　　　　　——基于異步優(yōu)化與 MyBatis-Plus 分頁(yè)插件思想的實(shí)踐

適用場(chǎng)景

數(shù)據(jù)量較大的單表分頁(yè)查詢
較復(fù)雜的多表關(guān)聯(lián)查詢，包含group by等無法進(jìn)行count優(yōu)化較耗時(shí)的分頁(yè)查詢

技術(shù)棧

核心框架：Spring Boot + MyBatis-Plus
異步編程：JDK 8+ 的 CompletableFuture
數(shù)據(jù)庫(kù)：MySQL 8.0
線程池：自定義線程池管理并行任務(wù)（如 ThreadPoolTaskExecutor）

實(shí)現(xiàn)思路

解決傳統(tǒng)分頁(yè)查詢中串行執(zhí)行 COUNT 與數(shù)據(jù)查詢的性能瓶頸，通過并行化減少總耗時(shí)，同時(shí)兼容復(fù)雜查詢場(chǎng)景（如多表關(guān)聯(lián)、DISTINCT 等）

兼容mybatisPlus分頁(yè)參數(shù)，復(fù)用 IPage 接口定義分頁(yè)參數(shù)（當(dāng)前頁(yè)、每頁(yè)條數(shù)），

借鑒 MyBatis-Plus 的 PaginationInnerInterceptor，通過實(shí)現(xiàn) MyBatis 的 Interceptor 接口，攔截 Executor#query 方法，動(dòng)態(tài)修改 SQL，

sql優(yōu)化適配：COUNT 優(yōu)化：自動(dòng)移除 ORDER BY，保留 GROUP BY 和 DISTINCT（需包裹子查詢），數(shù)據(jù)查詢：保留完整 SQL 邏輯，僅追加 LIMIT 和 OFFSET。

直接上代碼

使用簡(jiǎn)單

調(diào)用查詢方法前賦值page對(duì)象屬性total大于0數(shù)值則可進(jìn)入自定義分頁(yè)查詢方案。

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

線程池配置

@Configuration
public class ThreadPoolTaskExecutorConfig {

    public static final Integer CORE_POOL_SIZE = 20;
    public static final Integer MAX_POOL_SIZE = 40;
    public static final Integer QUEUE_CAPACITY = 200;
    public static final Integer KEEP_ALIVE_SECONDS = 60;

    @Bean("threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        //核心線程數(shù)
        threadPoolTaskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        //線程池最大線程數(shù)
        threadPoolTaskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        //隊(duì)列容量
        threadPoolTaskExecutor.setQueueCapacity(QUEUE_CAPACITY);
        //線程空閑存活時(shí)間
        threadPoolTaskExecutor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
        //線程前綴
        threadPoolTaskExecutor.setThreadNamePrefix("commonTask-");
        //拒絕策略
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //線程池初始化
        threadPoolTaskExecutor.initialize();

        return threadPoolTaskExecutor;
    }

    @Bean("countAsyncThreadPool")
    public ThreadPoolTaskExecutor getCountAsyncThreadPool() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        //核心線程數(shù)，根據(jù)負(fù)載動(dòng)態(tài)調(diào)整
        threadPoolTaskExecutor.setCorePoolSize(6);
        //線程池最大線程數(shù)，根據(jù)負(fù)載動(dòng)態(tài)調(diào)整
        threadPoolTaskExecutor.setMaxPoolSize(12);
        //隊(duì)列容量  隊(duì)列容量不宜過多，根據(jù)負(fù)載動(dòng)態(tài)調(diào)整
        threadPoolTaskExecutor.setQueueCapacity(2);
        //線程空閑存活時(shí)間
        threadPoolTaskExecutor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
        //線程前綴
        threadPoolTaskExecutor.setThreadNamePrefix("countAsync-");
        //拒絕策略  隊(duì)列滿時(shí)由調(diào)用者主線程執(zhí)行
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //線程池初始化
        threadPoolTaskExecutor.initialize();

        return threadPoolTaskExecutor;
    }
}

mybatis-plus配置類

@Configuration
@MapperScan("com.xxx.mapper")
public class MybatisPlusConfig {

    @Resource
    ThreadPoolTaskExecutor countAsyncThreadPool;
    @Resource
    ApplicationContext applicationContext;

    @Bean
    public MybatisPlusInterceptor mybatisPlusInterceptor() {
        MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
        interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL));
        return interceptor;
    }


    @Bean
    public PageParallelQueryInterceptor pageParallelQueryInterceptor() {
        PageParallelQueryInterceptor pageParallelQueryInterceptor = new PageParallelQueryInterceptor();
        pageParallelQueryInterceptor.setCountAsyncThreadPool(countAsyncThreadPool);
        pageParallelQueryInterceptor.setApplicationContext(applicationContext);
        return pageParallelQueryInterceptor;
    }
}

自定義mybatis攔截器

package com.example.dlock_demo.interceptor;

import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.statement.select.*;
import org.apache.ibatis.builder.StaticSqlSource;
import org.apache.ibatis.cache.CacheKey;
import org.apache.ibatis.executor.Executor;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.mapping.ResultMap;
import org.apache.ibatis.plugin.Interceptor;
import org.apache.ibatis.plugin.Intercepts;
import org.apache.ibatis.plugin.Invocation;
import org.apache.ibatis.plugin.Signature;
import org.apache.ibatis.session.*;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;


/**
 * Mybatis-分頁(yè)并行查詢攔截器
 *
 * @author shf
 */
@Intercepts({
        @Signature(type = Executor.class, method = "query",
                args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class}),
        @Signature(type = Executor.class, method = "query",
                args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class, CacheKey.class, BoundSql.class})
})
@Slf4j
public class PageParallelQueryInterceptor implements Interceptor {
    /**
     * 用于數(shù)據(jù)庫(kù)并行查詢線程池
     */
    private ThreadPoolTaskExecutor countAsyncThreadPool;
    /**
     * 容器上下文
     */
    private ApplicationContext applicationContext;

    private static final String LONG_RESULT_MAP_ID = "twoPhase-Long-ResultMap";
    private static final Map<String, MappedStatement> twoPhaseMsCache = new ConcurrentHashMap();

    public void setCountAsyncThreadPool(ThreadPoolTaskExecutor countAsyncThreadPool) {
        this.countAsyncThreadPool = countAsyncThreadPool;
    }

    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        Object[] args = invocation.getArgs();
        MappedStatement ms = (MappedStatement) args[0];
        Object parameter = args[1];

        //獲取分頁(yè)參數(shù)
        Page<?> page = getPageParameter(parameter);
        if (page == null || page.getSize() <= 0 || !page.searchCount() || page.getTotal() == 0) {
            return invocation.proceed();
        }
        //獲取Mapper方法(注解形式 需利用反射且只能應(yīng)用在mapper接口層，不推薦使用)
        /*Method method = getMapperMethod(ms);
        if (method == null || !method.isAnnotationPresent(PageParallelQuery.class)) {
            return invocation.proceed();
        }*/

        BoundSql boundSql = ms.getBoundSql(parameter);
        String originalSql = boundSql.getSql();
        //禁用mybatis plus PaginationInnerInterceptor count查詢
        page.setSearchCount(false);
        page.setTotal(0);
        args[2] = RowBounds.DEFAULT;
        CompletableFuture<Long> countFuture = resolveCountCompletableFuture(invocation, originalSql);
        //limit查詢
        long startTime = System.currentTimeMillis();
        Object proceed = invocation.proceed();
        log.info("原SQL數(shù)據(jù)查詢-耗時(shí)={}", System.currentTimeMillis() - startTime);
        page.setTotal(countFuture.get());

        return proceed;
    }

    private CompletableFuture<Long> resolveCountCompletableFuture(Invocation invocation, String originalSql) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                //查詢總條數(shù)
                long startTime = System.currentTimeMillis();
                long total = executeCountQuery(originalSql, invocation);
                log.info("分頁(yè)并行查詢COUNT總條數(shù)[{}]-耗時(shí)={}", total, System.currentTimeMillis() - startTime);
                return total;
            } catch (Throwable e) {
                log.error("page parallel query exception:", e);
                throw new CompletionException(e);
            }
        }, countAsyncThreadPool).exceptionally(throwable -> {
            log.error("page parallel query exception:", throwable);
            return 0L;
        });
    }

    private CompletableFuture<Object> resolveOriginalProceedCompletableFuture(Invocation invocation) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                long startTime = System.currentTimeMillis();
                Object proceed = invocation.proceed();
                log.info("原SQL數(shù)據(jù)查詢-耗時(shí)={}", System.currentTimeMillis() - startTime);
                return proceed;
            } catch (Throwable e) {
                throw new CompletionException(e);
            }
        }, countAsyncThreadPool).exceptionally(throwable -> {
            log.error("page parallel query original proceed exception:", throwable);
            return null;
        });
    }

    /**
     * 執(zhí)行count查詢
     */
    private long executeCountQuery(String originalSql, Invocation invocation)
            throws JSQLParserException, SQLException {

        //解析并修改SQL為count查詢
        Select countSelect = (Select) CCJSqlParserUtil.parse(originalSql);
        PlainSelect plainSelect = (PlainSelect) countSelect.getSelectBody();

        //修改select為count(*)
        /*plainSelect.setSelectItems(Collections.singletonList(
                new SelectExpressionItem(new Function("COUNT", new Column("*")))
        );*/
        // 移除排序和分頁(yè)
        Distinct distinct = plainSelect.getDistinct();
        GroupByElement groupBy = plainSelect.getGroupBy();
        String countSql = "";
        if (groupBy == null && distinct == null) {
            Expression countFuncExpression = CCJSqlParserUtil.parseExpression("COUNT(*)");
            plainSelect.setSelectItems(Collections.singletonList(
                    new SelectExpressionItem(countFuncExpression)));
            plainSelect.setOrderByElements(null);
            countSql = plainSelect.toString();
        } else if (groupBy != null) {
            plainSelect.setLimit(null);
            plainSelect.setOffset(null);
            countSql = "SELECT COUNT(*) FROM (" + plainSelect + ") TOTAL";
        } else {
            plainSelect.setOrderByElements(null);
            plainSelect.setLimit(null);
            plainSelect.setOffset(null);
            countSql = "SELECT COUNT(*) FROM (" + plainSelect + ") TOTAL";
        }
        //執(zhí)行count查詢
        return doCountQuery(invocation, countSql);
    }

    /**
     * 執(zhí)行修改后的COUNT(*)-SQL查詢
     */
    @SuppressWarnings("unchecked")
    private Long doCountQuery(Invocation invocation, String modifiedSql) {
        //Executor executor = (Executor) invocation.getTarget();
        //創(chuàng)建新會(huì)話（自動(dòng)獲取新連接）
        Executor executor;
        SqlSessionFactory sqlSessionFactory = applicationContext.getBean(SqlSessionFactory.class);
        try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.SIMPLE)) {
            //com.alibaba.druid.pool.DruidPooledConnection
            System.out.println("新會(huì)話Connection class: " + sqlSession.getConnection().getClass().getName());
            Field executorField = sqlSession.getClass().getDeclaredField("executor");
            executorField.setAccessible(true);
            executor = (Executor) executorField.get(sqlSession);

            Object[] args = invocation.getArgs();
            MappedStatement originalMs = (MappedStatement) args[0];
            Object parameter = args[1];
            //創(chuàng)建新的查詢參數(shù)
            Map<String, Object> newParameter = new HashMap<>();
            if (parameter instanceof Map) {
                // 復(fù)制原始參數(shù)但移除分頁(yè)參數(shù)
                Map<?, ?> originalParams = (Map<?, ?>) parameter;
                originalParams.forEach((k, v) -> {
                    if (!(v instanceof Page)) {
                        newParameter.put(k.toString(), v);
                    }
                });
            }
            //創(chuàng)建新的BoundSql
            BoundSql originalBoundSql = originalMs.getBoundSql(newParameter);
            BoundSql newBoundSql = new BoundSql(originalMs.getConfiguration(), modifiedSql, originalBoundSql.getParameterMappings(), newParameter);
            //復(fù)制原始參數(shù)值
            originalBoundSql.getParameterMappings().forEach(mapping -> {
                String prop = mapping.getProperty();
                if (mapping.getJavaType().isInstance(newParameter)) {
                    newBoundSql.setAdditionalParameter(prop, newParameter);
                } else if (newParameter instanceof Map) {
                    Object value = ((Map<?, ?>) newParameter).get(prop);
                    newBoundSql.setAdditionalParameter(prop, value);
                }
            });
            //創(chuàng)建新的BoundSql
            /*BoundSql originalBoundSql = originalMs.getBoundSql(parameter);
            BoundSql newBoundSql = new BoundSql(originalMs.getConfiguration(), modifiedSql,
                    originalBoundSql.getParameterMappings(), parameter);*/
            Configuration configuration = originalMs.getConfiguration();
            //創(chuàng)建臨時(shí)ResultMap
            ResultMap resultMap = new ResultMap.Builder(
                    configuration,
                    LONG_RESULT_MAP_ID,
                    //強(qiáng)制指定結(jié)果類型
                    Long.class,
                    //自動(dòng)映射列到簡(jiǎn)單類型
                    Collections.emptyList()
            ).build();
            if (!configuration.hasResultMap(LONG_RESULT_MAP_ID)) {
                configuration.addResultMap(resultMap);
            }

            String countMsId = originalMs.getId() + "_countMsId";
            MappedStatement mappedStatement = twoPhaseMsCache.computeIfAbsent(countMsId, (key) ->
                    this.getNewMappedStatement(modifiedSql, originalMs, newBoundSql, resultMap, countMsId));
            //執(zhí)行查詢
            List<Object> result = executor.query(mappedStatement, newParameter, RowBounds.DEFAULT, (ResultHandler<?>) args[3]);
            long total = 0L;
            if (CollectionUtils.isNotEmpty(result)) {
                Object o = result.get(0);
                if (o != null) {
                    total = Long.parseLong(o.toString());
                }
            }
            return total;
        } catch (Throwable e) {
            log.error("分頁(yè)并行查詢-executeCountQuery異常:", e);
        }
        return 0L;
    }

    private MappedStatement getNewMappedStatement(String modifiedSql, MappedStatement originalMs, BoundSql newBoundSql,
                                                  ResultMap resultMap, String msId) {
        //創(chuàng)建新的MappedStatement
        MappedStatement.Builder builder = new MappedStatement.Builder(
                originalMs.getConfiguration(),
                msId,
                new StaticSqlSource(originalMs.getConfiguration(), modifiedSql, newBoundSql.getParameterMappings()),
                originalMs.getSqlCommandType()
        );
        //復(fù)制重要屬性
        builder.resource(originalMs.getResource())
                .fetchSize(originalMs.getFetchSize())
                .timeout(originalMs.getTimeout())
                .statementType(originalMs.getStatementType())
                .keyGenerator(originalMs.getKeyGenerator())
                .keyProperty(originalMs.getKeyProperties() == null ? null : String.join(",", originalMs.getKeyProperties()))
                .resultMaps(resultMap == null ? originalMs.getResultMaps() : Collections.singletonList(resultMap))
                .parameterMap(originalMs.getParameterMap())
                .resultSetType(originalMs.getResultSetType())
                .cache(originalMs.getCache())
                .flushCacheRequired(originalMs.isFlushCacheRequired())
                .useCache(originalMs.isUseCache());
        return builder.build();
    }

    /**
     * 獲取分頁(yè)參數(shù)
     */
    private Page<?> getPageParameter(Object parameter) {
        if (parameter instanceof Map) {
            Map<?, ?> paramMap = (Map<?, ?>) parameter;
            return (Page<?>) paramMap.values().stream()
                    .filter(p -> p instanceof Page)
                    .findFirst()
                    .orElse(null);
        }
        return parameter instanceof Page ? (Page<?>) parameter : null;
    }

    /**
     * 獲取Mapper方法
     */
    private Method getMapperMethod(MappedStatement ms) {
        try {
            String methodName = ms.getId().substring(ms.getId().lastIndexOf(".") + 1);
            Class<?> mapperClass = Class.forName(ms.getId().substring(0, ms.getId().lastIndexOf(".")));
            return Arrays.stream(mapperClass.getMethods())
                    .filter(m -> m.getName().equals(methodName))
                    .findFirst()
                    .orElse(null);
        } catch (ClassNotFoundException e) {
            return null;
        }
    }
}

注意事項(xiàng)

優(yōu)先采用：建立合適的索引(如連接字段排序字段)、調(diào)整sql結(jié)構(gòu)(如減少子查詢)等方式優(yōu)化時(shí)效，若常規(guī)手段優(yōu)化后，因數(shù)據(jù)量較大且查詢條件較為復(fù)雜導(dǎo)致查詢依然慢尤其count查詢耗時(shí)，可以采用此方案一試。

有人可能會(huì)擔(dān)心并行查詢，在高并發(fā)場(chǎng)景可能會(huì)導(dǎo)致count查詢與limit數(shù)據(jù)查詢不一致，但其實(shí)只要沒有鎖，只要是分開的兩條sql查詢，原mybatisplus分頁(yè)插件也一樣面臨這個(gè)問題。

count優(yōu)化沒有進(jìn)行join語(yǔ)句判斷優(yōu)化，相當(dāng)于主動(dòng)關(guān)閉了page.setOptimizeJoinOfCountSql(false);因在一對(duì)多關(guān)聯(lián)等場(chǎng)景下可能會(huì)造成count查詢有誤，Mybatisplus官網(wǎng)也有相關(guān)提示，所以這里干脆舍棄了。

mybatisplus版本不同，可能會(huì)導(dǎo)致JsqlParser所使用的api有所不同，需要自己對(duì)應(yīng)版本修改下。本篇版本使用的3.5.1

關(guān)于線程池的線程數(shù)設(shè)置順便提一下：

網(wǎng)上流行一個(gè)說法：

1. CPU 密集型任務(wù)

特點(diǎn)：任務(wù)主要消耗 CPU 資源（如復(fù)雜計(jì)算、圖像處理）。

線程數(shù)建議：

核心線程數(shù)：CPU 核心數(shù) + 1（或等于CPU核心數(shù)，避免上下文切換過多）。
最大線程數(shù)：與核心線程數(shù)相同（防止過多線程競(jìng)爭(zhēng) CPU）。

2. I/O 密集型任務(wù)

特點(diǎn)：任務(wù)涉及大量等待（如網(wǎng)絡(luò)請(qǐng)求、數(shù)據(jù)庫(kù)讀寫）。

線程數(shù)建議：

核心線程數(shù)：2 * CPU 核心數(shù)（確保正常負(fù)載下的高效處理）。
最大線程數(shù)：根據(jù)系統(tǒng)資源調(diào)整（用于應(yīng)對(duì)突發(fā)高并發(fā)）。

其實(shí)這個(gè)說法來源于一個(gè)經(jīng)驗(yàn)公式推導(dǎo)而來：

threads = CPU核心數(shù) * (1 + 平均等待時(shí)間 / 平均計(jì)算時(shí)間)

《Java 虛擬機(jī)并發(fā)編程》中介紹

另一篇：《Java Concurrency in Practice》即《java并發(fā)編程實(shí)踐》，給出的線程池大小的估算公式：

Nthreads=Ncpu*Ucpu*(1+w/c)，其中 Ncpu=CPU核心數(shù)，Ucpu=cpu使用率，0~1；W/C=等待時(shí)間與計(jì)算時(shí)間的比率

仔細(xì)推導(dǎo)兩個(gè)公式，其實(shí)類似，在cpu使用率達(dá)100%時(shí)，其實(shí)結(jié)論是一致的，這時(shí)候計(jì)算線程數(shù)的公式就成了，Nthreads=Ncpu*100%*(1+w/c) =Ncpu*(1+w/c)。

那么在實(shí)踐應(yīng)用中計(jì)算的公式就出來了，【以下推算，不考慮內(nèi)存消耗等方面】，如下：

1、針對(duì)IO密集型，阻塞耗時(shí)w一般都是計(jì)算耗時(shí)幾倍c，假設(shè)阻塞耗時(shí)=計(jì)算耗時(shí)的情況下，Nthreads=Ncpu*(1+1)=2Ncpu,所以這種情況下，建議考慮2倍的CPU核心數(shù)做為線程數(shù)

2、對(duì)于計(jì)算密集型，阻塞耗時(shí)趨于0，即w/c趨于0，公式Nthreads = Ncpu。

實(shí)際應(yīng)用時(shí)要考慮同時(shí)設(shè)置了幾個(gè)隔離線程池，另外tomcat自帶的線程池也會(huì)共享宿主機(jī)公共資源。

posted @ 2025-04-27 15:56 字節(jié)悅動(dòng) 閱讀(786) 評(píng)論(0) 收藏舉報(bào)

刷新頁(yè)面返回頂部

北冥有魚