做服务端初步性能测试时，往往需要将一个方法顺序或并发运行 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;
- 当要并发存储的是有实例有状态的对象，且不适合在多线程中公共访问时，适合采用对象池。