计算一个服务端方法运行多次的平均耗时(Java)

做服务端初步性能测试时，往往需要将一个方法顺序或并发运行 N 次，计算最大、最小、平均耗时。

话不多说，上代码。

package sample.performance;

import java.util.function.Consumer;

/**
 * 消费器包装
 * Created by qinshu on 2021/7/11
 */
public class ConsumerWrapper<T> {

    private Consumer<T> consumer;
    private T param;

    public ConsumerWrapper(Consumer<T> consumer, T param) {
        this.consumer = consumer;
        this.param = param;
    }

    public void run() {
        consumer.accept(param);
    }
}

public class PerformanceTestFramework {

    public static final Integer DATA_SIZE = 100;
    public static final Integer RUNTIMES = 1;

    public static final Integer THREAD_NUMS = 1;

    static List<Long> costs = new CopyOnWriteArrayList<>();
    static int errorCount = 0;

    private static ThreadPoolExecutor executorService = new ThreadPoolExecutor(8, 8, 0L, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(1024));

    public static TestStrategyEnum SEQUENTIAL = TestStrategyEnum.SEQUENTIAL;
    public static TestStrategyEnum CONCURRENT = TestStrategyEnum.CONCURRENT;

    private static TestStrategy sequentialStrategy = new SeqTestStrategy();
    private static TestStrategy concurrentStrategy = new ConcurrentTestStrategy();

    public static void test(ConsumerWrapper consumerWrapper, TestStrategyEnum strategyEnum) {
        choose(strategyEnum).test(consumerWrapper);
    }

    private static TestStrategy choose(TestStrategyEnum strategyEnum) {
        return strategyEnum == TestStrategyEnum.SEQUENTIAL ? sequentialStrategy : concurrentStrategy;
    }

    enum TestStrategyEnum {
        SEQUENTIAL,
        CONCURRENT;
    }


    static class ConcurrentTestStrategy implements TestStrategy {

        @Override
        public void test(ConsumerWrapper consumerWrapper) {

            reset();

            CountDownLatch cdl = new CountDownLatch(RUNTIMES);
            for (int i=0; i < RUNTIMES; i++) {
                executorService.submit(() -> {
                    time(consumerWrapper);
                    cdl.countDown();
                });
            }
            try {
                cdl.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            executorService.shutdown();
            executorService.shutdownNow();

            report(this.getClass().getName());
        }
    }

    static class SeqTestStrategy implements TestStrategy {

        @Override
        public void test(ConsumerWrapper consumerWrapper) {

            reset();

            for (int i=0; i < RUNTIMES; i++) {
                time(consumerWrapper);
            }

            report(this.getClass().getName());
        }
    }

    interface TestStrategy {
        void test(ConsumerWrapper consumerWrapper);
    }

    private static void time(ConsumerWrapper consumerWrapper) {
        long start = System.currentTimeMillis();
        consumerWrapper.run();
        long end = System.currentTimeMillis();
        costs.add(end-start);
        System.out.println((end-start) + "ms");
    }

    private static void report(String className) {
        IntSummaryStatistics costStats = costs.stream().collect(Collectors.summarizingInt(Long::intValue));
        System.out.println(className + " Test: cost avg: " + costStats.getAverage() + " min: " + costStats.getMin() + " max: " + costStats.getMax() + " count: " + costStats.getCount());

        System.out.println(className + " Test: errorCount: " + errorCount);
    }

    private static void reset() {
        costs.clear();
        errorCount = 0;
    }
}

使用：

public class ThreatSqlQuery {

    public static void main(String[]args) {

        PreparedStatementWrapper preparedStatementWrapper = new PreparedStatementWrapper();

        try {

            String sql = "select p.pid, p.pname, f.fname, f.path from process_event as p inner join file as f on p.fname = f.fname inner join hash as h on f.hash = h.hash where p.eventId = 'Docker-Detect-Event-1626236000' and f.eventId = 'Docker-Detect-Event-1626236000'";

            PreparedStatement st = preparedStatementWrapper.create(sql);

            PerformanceTestFramework.test(
                    new ConsumerWrapper<>(sqlstr -> {
                        try {
                            preparedStatementWrapper.execute(st, new ArrayList<>());
                        } catch (SQLException ex) {
                            ex.printStackTrace();
                        }
                    }, sql), PerformanceTestFramework.SEQUENTIAL
            );

            st.close();

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

注意，这里直接使用 Statement 或 PreparedStatement 并发访问可能会有点问题，需要包装成线程安全的。

public class PreparedStatementWrapper {

    private Map<Integer, PreparedStatement> preparedStatementMap = new ConcurrentHashMap<>();

    private static CalciteConnection calciteConnection;

    static {

        try {
            Class.forName("org.apache.calcite.jdbc.Driver");

            Properties info = new Properties();
            info.setProperty("lex", "JAVA");

            Connection connection =
                    DriverManager.getConnection("jdbc:calcite:model=/Users/qinshu/workspace/vtdemo/src/main/resources/vt.json", info);

            calciteConnection =
                    connection.unwrap(CalciteConnection.class);

        } catch (Exception e1) {
            //
        }
    }

    public PreparedStatement create(String sql) throws SQLException {
        int hash = sql.hashCode();
        if (preparedStatementMap.get(hash) != null) {
            return preparedStatementMap.get(hash);
        }
        synchronized (sql) {
            preparedStatementMap.put(hash, calciteConnection.prepareStatement(sql));
            return preparedStatementMap.get(hash);
        }
    }

    public void execute(PreparedStatement st, List<PrepareStatementParamObject> params) throws SQLException {
        synchronized (st) {
            if (params != null) {
                for (PrepareStatementParamObject paramObject: params) {
                    st.setObject(paramObject.getIndex(), paramObject.getValue());
                }
            }
            ResultSet result = st.executeQuery();
            logResult(result);
            result.close();
        }
    }

    private static void logResult(ResultSet resultSet) {
        try {
            ResultSetMetaData metaData = resultSet.getMetaData();
            int colCount = metaData.getColumnCount();
            while(resultSet.next()) {
                for (int i=1; i <= colCount; i++) {
                    System.out.printf(resultSet.getString(i) + " ");
                }
                System.out.println();
            }
        } catch (Exception ex) {
            //
        }
    }
}

但是，PreparedStatement 使用 ConcurrentHashMap 是不合适的。因为 PreparedStatement 是线程不安全的，也不推荐在多线程里复用。更合适的方法是，采用对象池。

PrepareStatementFactory.java

/**
 * PrepareStatement对象工厂，和PrepareStatementPool是一对
 */
public class PrepareStatementFactory extends BaseKeyedPooledObjectFactory<String, PreparedStatement> {

    public final PreparedStatementWrapper preparedStatementWrapper;

    public PrepareStatementFactory(PreparedStatementWrapper preparedStatementWrapper) {
        this.preparedStatementWrapper = preparedStatementWrapper;
    }

    /**
     * 创建对象
     */
    @Override
    public PreparedStatement create(String sql) throws Exception {
        return preparedStatementWrapper.create(sql);
    }

    @Override
    public PooledObject<PreparedStatement> wrap(PreparedStatement preparedStatement) {
        return new DefaultPooledObject<>(preparedStatement);
    }

    /**
     * 验证对象是否有效
     */
    @Override
    public void activateObject(String sql, PooledObject<PreparedStatement> p) {
    }

    @Override
    public void passivateObject(String sql, PooledObject<PreparedStatement> p) {
    }

}

PrepareStatementPool.java

/**
 * PrepareStatement对象池
 */
public class PrepareStatementPool {

    private final PreparedStatementWrapper preparedStatementWrapper;
    private GenericKeyedObjectPool<String, PreparedStatement> objectPool;

    /**
     * 对象池每个key最大实例化对象数
     */
    private final static int TOTAL_PERKEY = 30;
    /**
     * 对象池每个key最大的闲置对象数
     */
    private final static int IDLE_PERKEY = 10;

    public PrepareStatementPool(PreparedStatementWrapper preparedStatementWrapper) {
        this.preparedStatementWrapper = preparedStatementWrapper;

        // 初始化对象池
        initObjectPool();
    }

    public PreparedStatementWrapper getPreparedStatementWrapper() {
        return preparedStatementWrapper;
    }

    /**
     * 初始化对象池
     */
    public void initObjectPool() {
        GenericKeyedObjectPoolConfig config = new GenericKeyedObjectPoolConfig();
        config.setMaxTotalPerKey(TOTAL_PERKEY);
        config.setMaxIdlePerKey(IDLE_PERKEY);
        config.setBlockWhenExhausted(true);
        KeyedPooledObjectFactory objectFactory = new PrepareStatementFactory(preparedStatementWrapper);
        if (objectPool == null) {
            objectPool = new GenericKeyedObjectPool<>(objectFactory, config);
        }
    }

    /**
     * 从对象池中获取对象
     */
    public PreparedStatement borrow(String sql) {
        try {
            return objectPool.borrowObject(sql);
        } catch (Exception ex) {
            throw new RuntimeException(ex.getCause());
        }
    }

    /**
     * 归还对象
     */
    public void returnObject(String sql, PreparedStatement p) {
        try {
            objectPool.returnObject(sql, p);
        } catch (Exception ex) {
            throw new RuntimeException(ex.getCause());
        }
    }

}

PreparedStatementWrapper.java

/**
 * 预编译语句缓存
 * NOTE: 应用方，需要作为@Component注入spring容器中
 */
public class PreparedStatementWrapper {

    private static final Logger logger = LoggerFactory.getLogger(PreparedStatementWrapper.class);
    private static final String LEX = "lex";
    private static final String CALCITE_DRIVER = "org.apache.calcite.jdbc.Driver";
    private static final String DEFAULT_URL = "jdbc:calcite:model=src/main/resources/virtualtable.json";
    private static final String ID = "id";

    /**
     * 创建连接
     */
    public CalciteConnection connection() {
        try {
            Class.forName(CALCITE_DRIVER);
            Properties info = new Properties();
            info.setProperty(LEX, Lex.JAVA.name()); //java类型，还有MYSQL、SQL_SERVER等类型
            Connection connection = DriverManager.getConnection(DEFAULT_URL, info);
            return connection.unwrap(CalciteConnection.class);
        } catch (Exception ex) {
            logger.error("PreparedStatementWrapper.connection error");
            throw new RuntimeException(ex);
        }
    }

    /**
     * 创建PreparedStatement对象
     */
    public PreparedStatement create(String sql) throws SQLException {
        return connection().prepareStatement(sql);
    }

    /**
     * 执行查询
     *
     * @param params 查询参数
     */
    public ResultSet execute(PreparedStatement preparedStatement, List<PrepareStatementParam> params) throws SQLException {
        if (params != null && !params.isEmpty()) {
            for (PrepareStatementParam paramObject : params) {
                // 赋值查询参数
                preparedStatement.setObject(paramObject.getIndex(), paramObject.getValue());
            }
        }
        // 执行查询
        ResultSet result = preparedStatement.executeQuery();
        preparedStatement.clearParameters();
        return result;
    }
}

使用：

public List<TableData> query(String sql, List<PrepareStatementParam> params) {
    // 1、从对象池中获取对象
    PreparedStatement preparedStatement = prepareStatementPool.borrow(sql);
    try {
        // 2、执行SQL
        ResultSet resultSet = preparedStatementWrapper.execute(preparedStatement, params);
        return preparedStatementWrapper.getResult(resultSet);
    } catch (SQLException ex) {
        logger.error("CalciteTemplate.query error，sql：{}", sql);
        throw new RuntimeException(ex);
    } finally {
        // 3、释放对象
        prepareStatementPool.returnObject(sql, preparedStatement);
    }
}

并发编程小提示：

• 当要并发地存储和访问无实例无状态单例的时候，适合使用 ConcurrentHashMap;
• 当要并发存储的是有实例有状态的对象，且不适合在多线程中公共访问时，适合采用对象池。