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

数据添加更改

master
lwj 2024-09-08 16:35:07 +08:00
parent 2c37501b9e
commit 288e36e2e7
1 changed files with 5 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -172,16 +172,12 @@ 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) {
// TableInfo tableInfoDataSources = tableInfoService.getById(basicId);
// Long basicId1 = tableInfoDataSources.getBasicId();
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();
HikariDataSource hikariDataSource = HikariPool.getHikariDataSource(dataSources); HikariDataSource hikariDataSource = HikariPool.getHikariDataSource(dataSources);
// HikariDataSource hikariDataSource = new HikariDataSource(hikariConfig);
ExecutorService executorService = Executors.newFixedThreadPool(8); ExecutorService executorService = Executors.newFixedThreadPool(8);
AtomicInteger addCount = new AtomicInteger(); AtomicInteger addCount = new AtomicInteger();
@ -192,11 +188,14 @@ public class ProductServiceImpl implements ProductService {
conn.setAutoCommit(false); // 开启事务 conn.setAutoCommit(false); // 开启事务
for (DataValue[][] batch : batches) { for (DataValue[][] batch : batches) {
log.info("数量batch{}", batch.length);
executorService.submit(() -> { executorService.submit(() -> {
try (Statement stmt = conn.createStatement()) { try (Statement stmt = conn.createStatement()) {
String sql = buildBatchInsertSQL(tableName, batch); String sql = buildBatchInsertSQL(tableName, batch);
stmt.executeUpdate(sql); stmt.executeUpdate(sql);
addCount.addAndGet(batch.length); addCount.addAndGet(batch.length);
} catch (SQLException e) { } catch (SQLException e) {
log.error("SQLException异常发生", e); log.error("SQLException异常发生", e);
@ -237,7 +236,6 @@ public class ProductServiceImpl implements ProductService {
// 删除最后一个逗号和空格 // 删除最后一个逗号和空格
columns.delete(columns.length() - 2, columns.length()); columns.delete(columns.length() - 2, columns.length());
// 构建值部分 // 构建值部分
for (DataValue[] dataValueList : batch) { for (DataValue[] dataValueList : batch) {
values.append("("); values.append("(");
@ -251,6 +249,7 @@ public class ProductServiceImpl implements ProductService {
values.delete(values.length() - 2, values.length()); values.delete(values.length() - 2, values.length());
// 完成 SQL 插入语句 // 完成 SQL 插入语句
String sql = "INSERT INTO " + tableName + " " + columns.toString() + ") " + values.toString(); String sql = "INSERT INTO " + tableName + " " + columns.toString() + ") " + values.toString();
log.info("执行的sql添加的sql语句"+sql);
return sql; return sql;
} }
@ -307,5 +306,4 @@ public class ProductServiceImpl implements ProductService {
} }