//示例代碼
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;
    }
}

@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;
    }
}

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;
        }
    }
}