Flink CDC并发压力测试数据完整性
·
package com.test;
import com.google.common.base.Joiner;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* Flink CDC压力测试工具：多线程批量增删改t_data，并记录操作日志到t_data_test_log
* */
public class TestCdc {
// ==================== 可配置参数 ==================== */
private static final int THREAD_COUNT = 5; // 自定义线程数
private static final int BATCH_SIZE = 10000; // 每批次操作记录数
private static final long BATCH_INTERVAL = 1000; // 批次间隔时间(ms)
private static final String JDBC_URL = "jdbc:postgresql://ip:2345/flink";
private static final String JDBC_USER = "";
private static final String JDBC_PWD = "";
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT); // 提交多线程任务
for (int i = 0; i < THREAD_COUNT; i++) {
String threadId = "THREAD_" + i;
// 必须用final修饰threadId，否则lambda中可能引用不到正确值
final String finalThreadId = threadId;
executor.submit(() -> {
int batchSeq = 0;
countDownLatch.countDown();
try {
countDownLatch.await();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
while (true) { // 无限循环持续执行
String batchNo = finalThreadId + "_BATCH_" + batchSeq++;
try {
Timestamp testTime = new Timestamp(System.currentTimeMillis());
executeBatch(testTime,batchNo, Thread.currentThread().getName());
// 打印执行日志，确认线程在运行
// System.out.printf("[%s] 批次[%s]执行完成，等待下一批次...%n",
// Thread.currentThread().getName(), batchNo);
TimeUnit.MILLISECONDS.sleep(BATCH_INTERVAL);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
//System.err.println("线程被中断: " + e.getMessage());
break;
} catch (Exception e) {
// 捕获所有异常，避免线程悄悄死掉
//System.err.printf("[%s] 批次[%s]执行失败: %s%n",
// Thread.currentThread().getName(), batchNo, e.getMessage());
// 失败后休眠3秒，减少报错频率
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
break;
}
}
}
});
}
// 移除shutdown和awaitTermination，线程池持续运行
// System.out.println("压力测试任务已启动，持续运行中（直接关进程停止）...");
// 主线程阻塞（可选，避免JVM意外退出，按需求决定是否保留）
synchronized (TestCdc.class) {
try {
TestCdc.class.wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
/**
* 单批次执行：新增→更新→删除 + 日志记录（同一事务）
*/
private static void executeBatch(Timestamp testTime,String batchNo, String threadId) {
Connection conn = null;
try {
// 获取数据库连接，关闭自动提交
conn = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PWD);
conn.setAutoCommit(false);
// 1. 批量新增数据 + 记录C类型日志
batchInsert(testTime,conn, batchNo, threadId);
conn.commit();
// 2. 批量更新数据（仅改param_value） + 记录U类型日志
// batchUpdate1(testTime,conn, batchNo, threadId);
// conn.commit();
// 3. 批量删除数据 + 记录D类型日志
//batchDelete1(testTime,conn, batchNo, threadId);
// 提交事务
conn.commit();
System.out.printf("[%s] 批次[%s] 执行成功%n", threadId, batchNo);
} catch (SQLException e) {
// 事务回滚
if (conn != null) {
try {
conn.rollback();
//System.err.printf("[%s] 批次[%s] 执行失败，已回滚: %s%n", threadId, batchNo, e.getMessage());
} catch (SQLException ex) {
//System.err.println("回滚失败: " + ex.getMessage());
}
}
} finally {
// 关闭连接
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
//System.err.println("关闭连接失败: " + e.getMessage());
}
}
}
}
/**
* 批量新增：插入t_data + 记录C类型日志到t_data_test_log
*/
private static void batchInsert(Timestamp testTime,Connection conn, String batchNo, String threadId) throws SQLException {
String insertSql = "INSERT INTO public.t_data_test (" +
"id, create_by, create_date, last_modified_by, last_modified_date, " +
"test_id, device_no, product_sn, mo_no, group_code, work_sn, " +
"param_code, param_value, upper, lower, test_time, " +
"load_time, test_emp, test_result, product_model, hcfr_id, temp, " +
"hum, datatype, rep_test_result, param_unit, area_id, org_id, " +
"is_deleted, tenant_id, param_name, device_name, group_name, workname, area_name, product_model_name" +
") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW(), NOW(), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
// 日志表插入SQL：对应t_data_test_log的4个字段（id, test_time, param_value, oper_type）
String logSql = "INSERT INTO public.t_data_test_log (id, test_time, param_value, oper_type) VALUES (?, ?, ?, ?)";
try (PreparedStatement psInsert = conn.prepareStatement(insertSql);
PreparedStatement psLog = conn.prepareStatement(logSql)) {
for (int i = 0; i < BATCH_SIZE; i++) {
long id =UUID.randomUUID().getMostSignificantBits();
String productSn = "SN_" + batchNo + "_" + i;
String paramValue = "INIT_VALUE_" + i;
//Timestamp testTime = new Timestamp(System.currentTimeMillis()); // 日志时间戳
// 设置主表插入参数（按字段顺序）
psInsert.setLong(1, id);
psInsert.setString(2, threadId);
psInsert.setTimestamp(3, testTime);
psInsert.setString(4, "test");
psInsert.setTimestamp(5, testTime);
psInsert.setNull(6, Types.NUMERIC);
psInsert.setString(7, "DEVICE_" + batchNo);
psInsert.setString(8, productSn);
psInsert.setString(9, "MO_" + batchNo);
psInsert.setString(10, "GROUP_" + batchNo);
psInsert.setString(11, "WS_" + batchNo);
psInsert.setString(12, "PARAM_" + i);
psInsert.setString(13, paramValue);
psInsert.setString(14, "UPPER");
psInsert.setString(15, "LOWER");
psInsert.setString(16, "TEST_EMP_" + threadId);
psInsert.setNull(17, Types.BIGINT);
psInsert.setString(18, "MODEL_" + batchNo);
psInsert.setNull(19, Types.NUMERIC);
psInsert.setString(20, "25℃");
psInsert.setString(21, "60%");
psInsert.setNull(22, Types.NUMERIC);
psInsert.setString(23, "N");
psInsert.setString(24, "UNIT");
psInsert.setNull(25, Types.BIGINT);
psInsert.setNull(26, Types.BIGINT);
psInsert.setInt(27, 0);
psInsert.setString(28, "TENANT_001");
psInsert.setString(29, "PARAM_NAME_" + i);
psInsert.setString(30, "DEVICE_NAME_" + batchNo);
psInsert.setString(31, "GROUP_NAME_" + batchNo);
psInsert.setString(32, "WS_NAME_" + batchNo);
psInsert.setString(33, "AREA_NAME_" + batchNo);
psInsert.setString(34, "MODEL_NAME_" + batchNo);
psInsert.addBatch();
// 设置日志表插入参数（C=新增）
psLog.setLong(1, id);
psLog.setTimestamp(2, testTime);
psLog.setString(3, paramValue);
psLog.setString(4, "C");
psLog.addBatch();
}
psInsert.executeBatch();
psLog.executeBatch();
}
}
private static void batchUpdate1(Timestamp testTime, Connection conn, String batchNo, String threadId) throws SQLException {
String selectIdsSql = "SELECT id FROM public.t_data_test WHERE product_sn LIKE ? LIMIT 200";
String updateSql = "UPDATE public.t_data_test SET param_value = ?, last_modified_by = ?, last_modified_date = NOW() WHERE id = ?";
String selectSql = "SELECT id, param_value FROM public.t_data_test WHERE id = ?";
String logSql = "INSERT INTO public.t_data_test_log (id, test_time, param_value, oper_type) VALUES (?, ?, ?, ?)";
String productSnLike = "SN_" + batchNo + "_%";
String newParamValue = "UPDATED_VALUE_" + batchNo;
// 1. 查询最多 200 条要更新的记录 ID（加行锁防止并发修改）
List<Long> ids = new ArrayList<>(200);
try (PreparedStatement psSelectIds = conn.prepareStatement(selectIdsSql)) {
psSelectIds.setString(1, productSnLike);
ResultSet rs = psSelectIds.executeQuery();
while (rs.next()) {
ids.add(rs.getLong("id"));
}
}
if (ids.isEmpty()) {
return; // 没有需要更新的记录
}
// 2. 批量更新（每次 200 条）
try (PreparedStatement psUpdate = conn.prepareStatement(updateSql)) {
for (Long id : ids) {
psUpdate.setString(1, newParamValue);
psUpdate.setString(2, threadId);
psUpdate.setLong(3, id);
psUpdate.addBatch();
}
psUpdate.executeBatch();
}
// 3. 查询更新后的数据并记录日志（U=修改）
try (PreparedStatement psSelect = conn.prepareStatement(selectSql);
PreparedStatement psLog = conn.prepareStatement(logSql)) {
for (Long id : ids) {
psSelect.setLong(1, id);
ResultSet rs = psSelect.executeQuery();
if (rs.next()) {
psLog.setLong(1, id);
psLog.setTimestamp(2, testTime);
psLog.setString(3, newParamValue);
psLog.setString(4, "U");
psLog.addBatch();
}
}
psLog.executeBatch();
}
}
/**
* 批量更新：修改t_data的param_value + 记录U类型日志到t_data_test_log
*/
private static void batchUpdate(Timestamp testTime,Connection conn, String batchNo, String threadId) throws SQLException {
String updateSql = "UPDATE public.t_data_test SET param_value = ?, last_modified_by = ?, last_modified_date = NOW() WHERE product_sn LIKE ?";
String selectSql = "SELECT id, param_value FROM public.t_data_test WHERE product_sn LIKE ?";
// 日志表插入SQL
String logSql = "INSERT INTO public.t_data_test_log (id, test_time, param_value, oper_type) VALUES (?, ?, ?, ?)";
String productSnLike = "SN_" + batchNo + "_%";
String newParamValue = "UPDATED_VALUE_" + batchNo;
// Timestamp testTime = new Timestamp(System.currentTimeMillis()); // 日志时间戳
// 1. 执行批量更新
try (PreparedStatement psUpdate = conn.prepareStatement(updateSql)) {
psUpdate.setString(1, newParamValue);
psUpdate.setString(2, threadId);
psUpdate.setString(3, productSnLike);
psUpdate.executeUpdate();
}
// 2. 查询更新后的数据并记录日志（U=修改）
try (PreparedStatement psSelect = conn.prepareStatement(selectSql);
PreparedStatement psLog = conn.prepareStatement(logSql)) {
psSelect.setString(1, productSnLike);
ResultSet rs = psSelect.executeQuery();
while (rs.next()) {
long id = rs.getLong("id");
// 记录更新后的param_value
psLog.setLong(1, id);
psLog.setTimestamp(2, testTime);
psLog.setString(3, newParamValue);
psLog.setString(4, "U");
psLog.addBatch();
}
psLog.executeBatch();
}
}
private static void batchDelete1(Timestamp testTime,Connection conn, String batchNo, String threadId) throws SQLException {
String productSnLike = "SN_" + batchNo + "_%";
//Timestamp testTime = new Timestamp(System.currentTimeMillis());
// 1. 先查询当前批次有多少数据，打印日志验证
String countSql = "SELECT COUNT(*) FROM public.t_data_test WHERE product_sn LIKE ?";
int totalBatchData = 0;
try (PreparedStatement psCount = conn.prepareStatement(countSql)) {
psCount.setString(1, productSnLike);
ResultSet rsCount = psCount.executeQuery();
if (rsCount.next()) {
totalBatchData = rsCount.getInt(1);
}
}
//System.out.printf("[%s] 批次[%s] 当前批次总数据量：%d 条%n", threadId, batchNo, totalBatchData);
if (totalBatchData == 0) {
// System.out.printf("[%s] 批次[%s] 无匹配数据，跳过删除%n", threadId, batchNo);
return;
}
// 1. 查询当前批次前200条待删除数据（用于日志）
String selectSql = "SELECT id, param_value FROM public.t_data_test " +
"WHERE product_sn LIKE ? LIMIT 200";
List<Long> deleteIds = new ArrayList<>();
//List<String> deleteParamValues = new ArrayList<>();
try (PreparedStatement psSelect = conn.prepareStatement(selectSql)) {
psSelect.setString(1, productSnLike);
ResultSet rs = psSelect.executeQuery();
while (rs.next()) {
deleteIds.add(rs.getLong("id"));
// deleteParamValues.add(rs.getString("param_value"));
}
}
if (deleteIds.isEmpty()) {
// System.out.printf("[%s] 批次[%s] 无待删除数据%n", threadId, batchNo);
return;
}
// 2. 记录删除日志
String logSql = "INSERT INTO public.t_data_test_log (id, test_time, param_value, oper_type) VALUES (?, ?, ?, ?)";
try (PreparedStatement psLog = conn.prepareStatement(logSql)) {
for (int i = 0; i < deleteIds.size(); i++) {
psLog.setLong(1, deleteIds.get(i));
psLog.setTimestamp(2, testTime);
psLog.setString(3, "");
psLog.setString(4, "D");
psLog.addBatch();
}
psLog.executeBatch();
}
// 3. 用LIMIT 200直接删除当前批次前200条（极简核心）
String deleteSql = "DELETE FROM public.t_data_test WHERE id in("+Joiner.on(",").join( deleteIds)+")";
String deleteLogSql = "DELETE FROM public.t_data_test_log WHERE oper_type in('C') and id in("+Joiner.on(",").join( deleteIds)+")";
try (PreparedStatement psDelete = conn.prepareStatement(deleteSql);
PreparedStatement psLogDelete = conn.prepareStatement(deleteLogSql)) {
int deletedCount = psDelete.executeUpdate();
int deletedCount2 = psLogDelete.executeUpdate();
// System.out.printf("[%s] 批次[%s] 成功删除前100条数据（实际删除%d条）%n",
// threadId, batchNo, deletedCount);
}
}
/**
* 批量删除：删除t_data数据 + 记录D类型日志到t_data_test_log
*/
private static void batchDelete(Connection conn, String batchNo, String threadId) throws SQLException {
String deleteSql = "DELETE FROM public.t_data_test WHERE product_sn LIKE ? ";
String selectSql = "SELECT id, param_value FROM public.t_data_test WHERE product_sn LIKE ?";
// 日志表插入SQL
String logSql = "INSERT INTO public.t_data_test_log (id, test_time, param_value, oper_type) VALUES (?, ?, ?, ?)";
String productSnLike = "SN_" + batchNo + "_%";
Timestamp testTime = new Timestamp(System.currentTimeMillis()); // 日志时间戳
// 1. 查询待删除数据并记录日志（D=删除）
try (PreparedStatement psSelect = conn.prepareStatement(selectSql);
PreparedStatement psLog = conn.prepareStatement(logSql)) {
psSelect.setString(1, productSnLike);
ResultSet rs = psSelect.executeQuery();
while (rs.next()) {
long id = rs.getLong("id");
String paramValue = rs.getString("param_value"); // 记录删除前的param_value
psLog.setLong(1, id);
psLog.setTimestamp(2, testTime);
psLog.setString(3, paramValue);
psLog.setString(4, "D");
psLog.addBatch();
}
psLog.executeBatch();
}
// 2. 执行批量删除
try (PreparedStatement psDelete = conn.prepareStatement(deleteSql)) {
psDelete.setString(1, productSnLike);
psDelete.executeUpdate();
}
}
}
更多推荐
1
0- 0
已为社区贡献3条内容
所有评论(0)