TreeData
/
cloud-etl-source
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

更新添加方法

master
lwj 2024-09-08 19:45:54 +08:00
parent 96718de9bd
commit fbea3645af
1 changed files with 304 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

454
cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/impl/ProductServiceImpl.java
View File

@ -14,8 +14,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.sql.Connection; import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException; import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
@ -169,177 +169,331 @@ public class ProductServiceImpl implements ProductService {
// } // }
// } // }
// 第二种
@Override // @Override
public int addProduct(Long basicId, Long tableId, DataValue[][] listList) { // public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
//
Source dataSources = sourceService.getById(basicId); // Source dataSources = sourceService.getById(basicId);
TableInfo tableInfo = tableInfoService.getById(tableId); // TableInfo tableInfo = tableInfoService.getById(tableId);
String tableName = tableInfo.getTableName(); // String tableName = tableInfo.getTableName();
//
HikariConfig hikariConfig = new HikariConfig(); // HikariConfig hikariConfig = new HikariConfig();
hikariConfig.setPoolName("HikariCP 连接池"); // hikariConfig.setPoolName("HikariCP 连接池");
hikariConfig.setDriverClassName(dataSources.getDriverName()); // hikariConfig.setDriverClassName(dataSources.getDriverName());
hikariConfig.setJdbcUrl(dataSources.getUrl()); // hikariConfig.setJdbcUrl(dataSources.getUrl());
hikariConfig.setUsername(dataSources.getUsername()); // hikariConfig.setUsername(dataSources.getUsername());
hikariConfig.setPassword(dataSources.getPassword()); // hikariConfig.setPassword(dataSources.getPassword());
hikariConfig.setMinimumIdle(2); // hikariConfig.setMinimumIdle(2);
hikariConfig.setMaximumPoolSize(10); // hikariConfig.setMaximumPoolSize(10);
//
HikariDataSource hikariDataSource = new HikariDataSource(hikariConfig); // HikariDataSource hikariDataSource = new HikariDataSource(hikariConfig);
//
//
// log.info("数据源ID的basicId{}值"+basicId);
// log.info("表的主键{}值"+tableId);
// for (DataValue[] dataValues : listList) {
// for (DataValue dataValue : dataValues) {
// log.info("里面的所有的值"+dataValue);
// }
// }
//
//// HikariDataSource hikariDataSource = HikariPool.getHikariDataSource(dataSources);
//
// ExecutorService executorService = Executors.newFixedThreadPool(8);
//
// AtomicInteger addCount = new AtomicInteger();
//
// // 分割数据为3000个批次
// List<DataValue[][]> batches = splitData(listList, 20);
//
// try (Connection conn = hikariDataSource.getConnection()) {
// conn.setAutoCommit(false); // 开启事务
//
// for (DataValue[][] batch : batches) {
// log.info("数量batch{}", batch.length);
//
// executorService.submit(() -> {
// try (Statement stmt = conn.createStatement()) {
//
// String sql = buildBatchInsertSQL(tableName, batch);
//
// stmt.executeUpdate(sql);
//
// addCount.addAndGet(batch.length);
// } catch (SQLException e) {
// log.error("SQLException异常发生", e);
// try {
// conn.rollback(); // 回滚事务
// } catch (SQLException ex) {
// log.error("回滚事务失败", ex);
// throw new RuntimeException(ex);
// }
// throw new RuntimeException(e);
// }
// });
// }
//
// executorService.shutdown();
// executorService.awaitTermination(1, TimeUnit.HOURS);
// conn.commit(); // 提交事务
// } catch (InterruptedException e) {
// Thread.currentThread().interrupt();
// throw new RuntimeException(e);
// } catch (SQLException e) {
// throw new RuntimeException(e);
// } finally {
// close(hikariDataSource); // 关闭数据源
// }
//
// return addCount.get();
// }
//
// private String buildBatchInsertSQL(String tableName, DataValue[][] batch) {
//
//
// for (DataValue[] dataValues : batch) {
// for (DataValue dataValue : dataValues) {
// log.info("我的键{}，值{}", dataValues, dataValue);
// }
// }
//
// StringBuilder columns = new StringBuilder("(");
// StringBuilder values = new StringBuilder("VALUES ");
// // 构建字段名
// for (DataValue dataValue : batch[0]) {
// String key = dataValue.getKey();
// columns.append(key).append(", ");
// }
//
// // 删除最后一个逗号和空格
// columns.delete(columns.length() - 2, columns.length());
// log.info("陈思豪是傻逼"+columns);
// int count=0;
// // 构建值部分
// for (DataValue[] dataValueList : batch) {
// values.append("(");
// for (DataValue dataValue : dataValueList) {
// Object value = dataValue.getValue();
// values.append(formatValue(dataValue.getType(), value)).append(", ");
// log.info("值{{{{aaaaaa}}}}}}}}}"+count+"+个数+"+value);
// }
// values.delete(values.length() - 2, values.length());
// values.append("), ");
// }
//
//
// values.delete(values.length() - 2, values.length());
// log.info("值++++++++++++++++++++++++++++++++++++++++++++++++++++++++"+values.toString());
// // 完成 SQL 插入语句
// String sql = "INSERT INTO " + tableName + " " + columns.toString() + ") " + values.toString();
// log.info("执行的sql添加的sql语句"+sql);
// return sql;
// }
//
// private String formatValue(DataType type, Object value) {
//
// if (type == DataType.VARCHAR || type == DataType.TEXT) {
//
// return "'" + value.toString().replace("'", "''") + "'";
// } else if (type == DataType.BIGINT) {
//
// return value.toString();
// } else if (type == DataType.INT) {
//
// return value.toString();
// } else if (type == DataType.DECIMAL) {
//
// return value.toString();
// } else if (type == DataType.DATETIME) {
//
// return "'" + new java.sql.Date(((Date) value).getTime()) + "'";
// } else if (type == DataType.DOUBLE) {
//
// return value.toString();
// } else {
// // 其他类型的处理
//
// return "'" + value.toString() + "'";
// }
//
// }
//
// private List<DataValue[][]> splitData(DataValue[][] listList, int batchSize) {
// List<DataValue[][]> batches = new ArrayList<>();
// int totalSize = listList.length;
// int numBatches = (int) Math.ceil((double) totalSize / batchSize);
//
// for (int i = 0; i < numBatches; i++) {
// int start = i * batchSize;
// int end = Math.min(start + batchSize, totalSize);
// DataValue[][] batch = new DataValue[end - start][];
// System.arraycopy(listList, start, batch, 0, end - start);
// batches.add(batch);
// }
//
// log.info("");
// return batches;
// }
//
// private static void close(HikariDataSource dataSource) {
// if (dataSource != null) {
// dataSource.close();
// }
// }
log.info("数据源ID的basicId{}值"+basicId); //第三种
log.info("表的主键{}值"+tableId);
for (DataValue[] dataValues : listList) {
for (DataValue dataValue : dataValues) { public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
log.info("里面的所有的值"+dataValue); Source dataSources = sourceService.getById(basicId);
TableInfo tableInfo = tableInfoService.getById(tableId);
String tableName = tableInfo.getTableName();
HikariConfig hikariConfig = new HikariConfig();
hikariConfig.setPoolName("HikariCP 连接池");
hikariConfig.setDriverClassName(dataSources.getDriverName());
hikariConfig.setJdbcUrl(dataSources.getUrl());
hikariConfig.setUsername(dataSources.getUsername());
hikariConfig.setPassword(dataSources.getPassword());
hikariConfig.setMinimumIdle(2);
hikariConfig.setMaximumPoolSize(10);
HikariDataSource hikariDataSource = new HikariDataSource(hikariConfig);
log.info("数据源ID的basicId值: {}", basicId);
log.info("表的主键值: {}", tableId);
for (DataValue[] dataValues : listList) {
for (DataValue dataValue : dataValues) {
log.info("里面的所有的值: {}", dataValue);
}
} }
}
// HikariDataSource hikariDataSource = HikariPool.getHikariDataSource(dataSources); ExecutorService executorService = Executors.newFixedThreadPool(8);
AtomicInteger addCount = new AtomicInteger();
ExecutorService executorService = Executors.newFixedThreadPool(8); // 分割数据为3000个批次
List<DataValue[][]> batches = splitData(listList, 20);
AtomicInteger addCount = new AtomicInteger(); try (Connection conn = hikariDataSource.getConnection()) {
conn.setAutoCommit(false); // 开启事务
// 分割数据为3000个批次 for (DataValue[][] batch : batches) {
List<DataValue[][]> batches = splitData(listList, 20); log.info("数量batch: {}", batch.length);
try (Connection conn = hikariDataSource.getConnection()) { executorService.submit(() -> {
conn.setAutoCommit(false); // 开启事务 try (PreparedStatement stmt = conn.prepareStatement(buildBatchInsertSQL(tableName, batch))) {
int index = 1;
for (DataValue[][] batch : batches) { for (DataValue[] dataValues : batch) {
log.info("数量batch{}", batch.length); for (DataValue dataValue : dataValues) {
stmt.setObject(index++, dataValue.getValue());
executorService.submit(() -> { }
try (Statement stmt = conn.createStatement()) { stmt.addBatch();
}
String sql = buildBatchInsertSQL(tableName, batch); stmt.executeBatch();
addCount.addAndGet(batch.length);
stmt.executeUpdate(sql); } catch (SQLException e) {
log.error("SQLException异常发生", e);
addCount.addAndGet(batch.length); try {
} catch (SQLException e) { conn.rollback(); // 回滚事务
log.error("SQLException异常发生", e); } catch (SQLException ex) {
try { log.error("回滚事务失败", ex);
conn.rollback(); // 回滚事务 throw new RuntimeException(ex);
} catch (SQLException ex) { }
log.error("回滚事务失败", ex); throw new RuntimeException(e);
throw new RuntimeException(ex);
} }
throw new RuntimeException(e); });
} }
});
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.HOURS);
conn.commit(); // 提交事务
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
} catch (SQLException e) {
throw new RuntimeException(e);
} finally {
close(hikariDataSource); // 关闭数据源
} }
executorService.shutdown(); return addCount.get();
executorService.awaitTermination(1, TimeUnit.HOURS);
conn.commit(); // 提交事务
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
} catch (SQLException e) {
throw new RuntimeException(e);
} finally {
close(hikariDataSource); // 关闭数据源
} }
return addCount.get(); private String buildBatchInsertSQL(String tableName, DataValue[][] batch) {
} StringBuilder columns = new StringBuilder("(");
StringBuilder values = new StringBuilder("VALUES ");
private String buildBatchInsertSQL(String tableName, DataValue[][] batch) { // 构建字段名
for (DataValue dataValue : batch[0]) {
String key = dataValue.getKey();
for (DataValue[] dataValues : batch) { columns.append(key).append(", ");
for (DataValue dataValue : dataValues) {
log.info("我的键{}，值{}", dataValues, dataValue);
} }
}
StringBuilder columns = new StringBuilder("("); // 删除最后一个逗号和空格
StringBuilder values = new StringBuilder("VALUES "); columns.delete(columns.length() - 2, columns.length());
// 构建字段名
for (DataValue dataValue : batch[0]) {
String key = dataValue.getKey();
columns.append(key).append(", ");
}
// 删除最后一个逗号和空格 // 构建值部分
columns.delete(columns.length() - 2, columns.length()); for (DataValue[] dataValueList : batch) {
log.info("陈思豪是傻逼"+columns); values.append("(");
int count=0; for (DataValue dataValue : dataValueList) {
// 构建值部分 Object value = dataValue.getValue();
for (DataValue[] dataValueList : batch) { values.append("?, "); // 使用占位符
values.append("("); }
for (DataValue dataValue : dataValueList) { values.delete(values.length() - 2, values.length());
Object value = dataValue.getValue(); values.append("), ");
values.append(formatValue(dataValue.getType(), value)).append(", ");
log.info("值{{{{aaaaaa}}}}}}}}}"+count+"+个数+"+value);
} }
values.delete(values.length() - 2, values.length()); values.delete(values.length() - 2, values.length());
values.append("), "); // 完成 SQL 插入语句
String sql = "INSERT INTO " + tableName + " " + columns.toString() + " " + values.toString() + ")";
log.info("执行的sql添加的sql语句: {}", sql);
return sql;
}
private String formatValue(DataType type, Object value) {
if (type == DataType.VARCHAR || type == DataType.TEXT) {
return "'" + value.toString().replace("'", "''") + "'";
} else if (type == DataType.BIGINT) {
return value.toString();
} else if (type == DataType.INT) {
return value.toString();
} else if (type == DataType.DECIMAL) {
return value.toString();
} else if (type == DataType.DATETIME) {
return "'" + new java.sql.Date(((Date) value).getTime()) + "'";
} else if (type == DataType.DOUBLE) {
return value.toString();
} else {
// 其他类型的处理
return "'" + value.toString() + "'";
}
}
private List<DataValue[][]> splitData(DataValue[][] listList, int batchSize) {
List<DataValue[][]> batches = new ArrayList<>();
int totalSize = listList.length;
int numBatches = (int) Math.ceil((double) totalSize / batchSize);
for (int i = 0; i < numBatches; i++) {
int start = i * batchSize;
int end = Math.min(start + batchSize, totalSize);
DataValue[][] batch = new DataValue[end - start][];
System.arraycopy(listList, start, batch, 0, end - start);
batches.add(batch);
}
log.info("");
return batches;
}
private static void close(HikariDataSource dataSource) {
if (dataSource != null) {
dataSource.close();
}
} }
values.delete(values.length() - 2, values.length());
log.info("值++++++++++++++++++++++++++++++++++++++++++++++++++++++++"+values.toString());
// 完成 SQL 插入语句
String sql = "INSERT INTO " + tableName + " " + columns.toString() + ") " + values.toString();
log.info("执行的sql添加的sql语句"+sql);
return sql;
}
private String formatValue(DataType type, Object value) {
if (type == DataType.VARCHAR || type == DataType.TEXT) {
return "'" + value.toString().replace("'", "''") + "'";
} else if (type == DataType.BIGINT) {
return value.toString();
} else if (type == DataType.INT) {
return value.toString();
} else if (type == DataType.DECIMAL) {
return value.toString();
} else if (type == DataType.DATETIME) {
return "'" + new java.sql.Date(((Date) value).getTime()) + "'";
} else if (type == DataType.DOUBLE) {
return value.toString();
} else {
// 其他类型的处理
return "'" + value.toString() + "'";
}
}
private List<DataValue[][]> splitData(DataValue[][] listList, int batchSize) {
List<DataValue[][]> batches = new ArrayList<>();
int totalSize = listList.length;
int numBatches = (int) Math.ceil((double) totalSize / batchSize);
for (int i = 0; i < numBatches; i++) {
int start = i * batchSize;
int end = Math.min(start + batchSize, totalSize);
DataValue[][] batch = new DataValue[end - start][];
System.arraycopy(listList, start, batch, 0, end - start);
batches.add(batch);
}
log.info("");
return batches;
}
private static void close(HikariDataSource dataSource) {
if (dataSource != null) {
dataSource.close();
}
}