third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: performance test for h2

Browse Source
This commit is contained in:
tjq 2024-02-05 00:12:37 +08:00
parent ff84d46713
commit 1ba74bf0af
5 changed files with 181 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
powerjob-worker/src/main/java/tech/powerjob/worker/persistence/TaskDAOImpl.java
View File

@ -17,11 +17,20 @@ import java.util.Map;
* @author tjq
* @since 2020/3/17
*/
@AllArgsConstructor
public class TaskDAOImpl implements TaskDAO {
private final boolean useIndex;
private final ConnectionFactory connectionFactory;
public TaskDAOImpl(ConnectionFactory connectionFactory) {
this(false, connectionFactory);
}
public TaskDAOImpl(boolean useIndex, ConnectionFactory connectionFactory) {
this.useIndex = useIndex;
this.connectionFactory = connectionFactory;
}
@Override
public void initTable() throws Exception {
@ -30,9 +39,14 @@ public class TaskDAOImpl implements TaskDAO {
// bigint(20) 与 Java Long 取值范围完全一致
String createTableSQL = "create table task_info (task_id varchar(255), instance_id bigint, sub_instance_id bigint, task_name varchar(255), task_content blob, address varchar(255), status int, result text, failed_cnt int, created_time bigint, last_modified_time bigint, last_report_time bigint, constraint pkey unique (instance_id, task_id))";
String createIndexSQL = "create INDEX idx_status ON task_info (status)";
try (Connection conn = connectionFactory.getConnection(); Statement stat = conn.createStatement()) {
stat.execute(delTableSQL);
stat.execute(createTableSQL);
if (useIndex) {
stat.execute(createIndexSQL);
}
}
}

31
powerjob-worker/src/test/java/tech/powerjob/worker/persistence/AbstractTaskDAOTest.java Normal file
View File

@ -0,0 +1,31 @@
package tech.powerjob.worker.persistence;
import tech.powerjob.worker.common.constants.TaskStatus;
import java.nio.charset.StandardCharsets;
/**
* AbstractTaskDAOTest
*
* @author tjq
* @since 2024/2/4
*/
public class AbstractTaskDAOTest {
protected static TaskDO buildTaskDO(String taskId, Long instanceId, TaskStatus taskStatus) {
TaskDO taskDO = new TaskDO();
taskDO.setTaskId(taskId);
taskDO.setInstanceId(instanceId);
taskDO.setSubInstanceId(instanceId);
taskDO.setTaskName("TEST_TASK");
taskDO.setTaskContent("TEST_CONTENT".getBytes(StandardCharsets.UTF_8));
taskDO.setAddress("127.0.0.1:10086");
taskDO.setStatus(taskStatus.getValue());
taskDO.setResult("SUCCESS");
taskDO.setFailedCnt(0);
taskDO.setLastModifiedTime(System.currentTimeMillis());
taskDO.setLastReportTime(System.currentTimeMillis());
taskDO.setCreatedTime(System.currentTimeMillis());
return taskDO;
}
}

111
powerjob-worker/src/test/java/tech/powerjob/worker/persistence/TaskDAOPerformanceTest.java Normal file
View File

@ -0,0 +1,111 @@
package tech.powerjob.worker.persistence;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.util.StopWatch;
import tech.powerjob.worker.common.constants.StoreStrategy;
import tech.powerjob.worker.common.constants.TaskStatus;
import tech.powerjob.worker.core.processor.TaskResult;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicLong;
/**
* 任务持久化层 - 性能测试
*
* @author tjq
* @since 2024/2/4
*/
@Slf4j(topic = "PERFORMANCE_TEST_LOGGER")
public class TaskDAOPerformanceTest extends AbstractTaskDAOTest {
private static final int INSERT_NUM = 100000;
private static final Long INSTANCE_ID = 10086L;
@Test
void testInsert() throws Exception {
TaskDAO noIndexDao = initTaskDao(false);
TaskDAO indexDao = initTaskDao(true);
for (int i = 0; i < 1; i++) {
testWriteThenRead(noIndexDao, INSERT_NUM, "no-idx-" + i);
testWriteThenRead(indexDao, INSERT_NUM, "uu-idx-" + i);
}
}
@SneakyThrows
private void testWriteThenRead(TaskDAO taskDAO, int num, String taskName) {
String logKey = "testWriteThenRead-" + taskName;
StopWatch stopWatch = new StopWatch();
AtomicLong atomicLong = new AtomicLong();
ForkJoinPool pool = new ForkJoinPool(256);
CountDownLatch latch = new CountDownLatch(num);
stopWatch.start("Insert");
for (int i = 0; i < num; i++) {
pool.execute(() -> {
long id = atomicLong.incrementAndGet();
String taskId = String.format("%s.%d", taskName, id);
TaskDO taskDO = buildTaskDO(taskId, INSTANCE_ID, TaskStatus.of(ThreadLocalRandom.current().nextInt(1, 7)));
try {
long s = System.currentTimeMillis();
taskDAO.save(taskDO);
long cost = System.currentTimeMillis() - s;
if (cost > 10) {
log.warn("[{}] id={} save cost too much: {}", logKey, id, cost);
}
} catch (Exception e) {
log.error("[{}] id={} save failed!", logKey, id, e);
} finally {
latch.countDown();
}
});
}
latch.await();
stopWatch.stop();
stopWatch.start("READ-getAllTaskResult");
// 测试读
List<TaskResult> allTaskResult = taskDAO.getAllTaskResult(INSTANCE_ID, INSTANCE_ID);
stopWatch.stop();
// 测试统计
stopWatch.start("READ-countByStatus");
SimpleTaskQuery query = new SimpleTaskQuery();
query.setInstanceId(INSTANCE_ID);
query.setSubInstanceId(INSTANCE_ID);
query.setQueryContent("status, count(*) as num");
query.setOtherCondition("GROUP BY status");
List<Map<String, Object>> countByStatus = taskDAO.simpleQueryPlus(query);
stopWatch.stop();
String prettyPrint = stopWatch.prettyPrint();
System.out.println(logKey + ": " + prettyPrint);
log.info("[{}] {}", logKey, prettyPrint);
}
@SneakyThrows
private TaskDAO initTaskDao(boolean useIndex) {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.initDatasource(StoreStrategy.DISK);
TaskDAO taskDAO = new TaskDAOImpl(useIndex, connectionFactory);
taskDAO.initTable();
return taskDAO;
}
}

20
powerjob-worker/src/test/java/tech/powerjob/worker/persistence/TaskDAOTest.java
View File

@ -24,7 +24,7 @@ import static org.junit.jupiter.api.Assertions.*;
* @since 2022/10/23
*/
@Slf4j
class TaskDAOTest {
class TaskDAOTest extends AbstractTaskDAOTest {
private static TaskDAO taskDAO;
@ -94,22 +94,4 @@ class TaskDAOTest {
assert allTaskResult.size() == 2;
}
private static TaskDO buildTaskDO(String taskId, Long instanceId, TaskStatus taskStatus) {
TaskDO taskDO = new TaskDO();
taskDO.setTaskId(taskId);
taskDO.setInstanceId(instanceId);
taskDO.setSubInstanceId(instanceId);
taskDO.setTaskName("TEST_TASK");
taskDO.setTaskContent("TEST_CONTENT".getBytes(StandardCharsets.UTF_8));
taskDO.setAddress("127.0.0.1:10086");
taskDO.setStatus(taskStatus.getValue());
taskDO.setResult("SUCCESS");
taskDO.setFailedCnt(0);
taskDO.setLastModifiedTime(System.currentTimeMillis());
taskDO.setLastReportTime(System.currentTimeMillis());
taskDO.setCreatedTime(System.currentTimeMillis());
return taskDO;
}
}

22
powerjob-worker/src/test/resources/logback-test.xml
View File

@ -20,6 +20,28 @@
<appender-ref ref="STDOUT"/>
</logger>
<!-- 性能测试专用，输出到本地文件，防止日志 IO 本身成为瓶颈 -->
<appender name="PERFORMANCE_TEST_APPENDER" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${user.home}/powerjob/worker/test/logs/performance_test.log</file>
<encoder>
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS}|%thread|%msg%n</pattern>
<charset>UTF-8</charset>
</encoder>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<fileNamePattern>${user.home}/powerjob/worker/test/logs/performance_test.log.%d{yyyy-MM-dd}</fileNamePattern>
<MaxHistory>7</MaxHistory>
</rollingPolicy>
</appender>
<appender name="ASYNC_PERFORMANCE_TEST_APPENDER" class="ch.qos.logback.classic.AsyncAppender">
<queueSize>512</queueSize>
<discardingThreshold>0</discardingThreshold>
<neverBlock>true</neverBlock>
<appender-ref ref="PERFORMANCE_TEST_APPENDER"/>
</appender>
<logger name="PERFORMANCE_TEST_LOGGER" level="INFO" additivity="false">
<appender-ref ref="ASYNC_PERFORMANCE_TEST_APPENDER"/>
</logger>
<!-- 控制台输出日志级别 -->
<root level="INFO">
<appender-ref ref="STDOUT"/>