lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

DataJdbcSource优化

Browse Source
This commit is contained in:
redkale
2023-07-24 22:51:35 +08:00
parent f33a1b997f
commit 3c45093632
3 changed files with 59 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/main/java/org/redkale/source/AbstractDataSource.java
View File

@@ -1276,6 +1276,12 @@ public abstract class AbstractDataSource extends AbstractService implements Data
protected DefaultDataBatch() { protected DefaultDataBatch() {
} }
public DataBatch run(Runnable task) {
Objects.requireNonNull(task);
this.actions.add(new RunnableBatchAction(task));
return this;
}
@Override //entitys不一定是同一表的数据 @Override //entitys不一定是同一表的数据
public <T> DataBatch insert(T... entitys) { public <T> DataBatch insert(T... entitys) {
for (T t : entitys) { for (T t : entitys) {
@@ -1429,6 +1435,15 @@ public abstract class AbstractDataSource extends AbstractService implements Data
} }
protected static class RunnableBatchAction extends BatchAction {
public Runnable task;
public RunnableBatchAction(Runnable task) {
this.task = task;
}
}
protected static class InsertBatchAction1 extends BatchAction { protected static class InsertBatchAction1 extends BatchAction {
public Object entity; public Object entity;

4
src/main/java/org/redkale/source/DataBatch.java
View File

@@ -4,7 +4,7 @@
package org.redkale.source; package org.redkale.source;
import java.io.Serializable; import java.io.Serializable;
import org.redkale.util.*; import org.redkale.util.SelectColumn;
/** /**
* DataSource批量操作对象，操作类型只能是增删改 <br> * DataSource批量操作对象，操作类型只能是增删改 <br>
@@ -22,6 +22,8 @@ public interface DataBatch {
return new AbstractDataSource.DefaultDataBatch(); return new AbstractDataSource.DefaultDataBatch();
} }
public DataBatch run(Runnable task);
public <T> DataBatch insert(T... entitys); public <T> DataBatch insert(T... entitys);
public <T> DataBatch delete(T... entitys); public <T> DataBatch delete(T... entitys);

138
src/main/java/org/redkale/source/DataJdbcSource.java
View File

@@ -218,11 +218,16 @@ public class DataJdbcSource extends AbstractDataSqlSource {
SourceConnection conn = null; SourceConnection conn = null;
try { try {
conn = writePool.pollTransConnection(); conn = writePool.pollTransConnection();
conn.setAutoCommit(false);
for (BatchAction action : dataBatch.actions) { for (BatchAction action : dataBatch.actions) {
if (action instanceof InsertBatchAction1) { if (action instanceof RunnableBatchAction) {
RunnableBatchAction act = (RunnableBatchAction) action;
act.task.run();
} else if (action instanceof InsertBatchAction1) {
InsertBatchAction1 act = (InsertBatchAction1) action; InsertBatchAction1 act = (InsertBatchAction1) action;
EntityInfo info = apply(act.entity.getClass()); EntityInfo info = apply(act.entity.getClass());
c += insertDB(true, conn, info, act.entity); c += insertDBStatement(true, conn, info, act.entity);
} else if (action instanceof DeleteBatchAction1) { } else if (action instanceof DeleteBatchAction1) {
DeleteBatchAction1 act = (DeleteBatchAction1) action; DeleteBatchAction1 act = (DeleteBatchAction1) action;
@@ -231,7 +236,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
Map<String, List<Serializable>> pkmap = info.getTableMap(pk); Map<String, List<Serializable>> pkmap = info.getTableMap(pk);
String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]); String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]);
String[] sqls = deleteSql(info, pkmap); String[] sqls = deleteSql(info, pkmap);
c += deleteDB(true, conn, info, tables, null, null, pkmap, sqls); c += deleteDBStatement(true, conn, info, tables, null, null, pkmap, sqls);
} else if (action instanceof DeleteBatchAction2) { } else if (action instanceof DeleteBatchAction2) {
DeleteBatchAction2 act = (DeleteBatchAction2) action; DeleteBatchAction2 act = (DeleteBatchAction2) action;
@@ -239,37 +244,37 @@ public class DataJdbcSource extends AbstractDataSqlSource {
Map<String, List<Serializable>> pkmap = info.getTableMap(act.pk); Map<String, List<Serializable>> pkmap = info.getTableMap(act.pk);
String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]); String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]);
String[] sqls = deleteSql(info, pkmap); String[] sqls = deleteSql(info, pkmap);
c += deleteDB(true, conn, info, tables, null, null, pkmap, sqls); c += deleteDBStatement(true, conn, info, tables, null, null, pkmap, sqls);
} else if (action instanceof DeleteBatchAction3) { } else if (action instanceof DeleteBatchAction3) {
DeleteBatchAction3 act = (DeleteBatchAction3) action; DeleteBatchAction3 act = (DeleteBatchAction3) action;
EntityInfo info = apply(act.clazz); EntityInfo info = apply(act.clazz);
String[] tables = info.getTables(act.node); String[] tables = info.getTables(act.node);
String[] sqls = deleteSql(info, tables, act.flipper, act.node); String[] sqls = deleteSql(info, tables, act.flipper, act.node);
c += deleteDB(true, conn, info, tables, act.flipper, act.node, null, sqls); c += deleteDBStatement(true, conn, info, tables, act.flipper, act.node, null, sqls);
} else if (action instanceof UpdateBatchAction1) { } else if (action instanceof UpdateBatchAction1) {
UpdateBatchAction1 act = (UpdateBatchAction1) action; UpdateBatchAction1 act = (UpdateBatchAction1) action;
EntityInfo info = apply(act.entity.getClass()); EntityInfo info = apply(act.entity.getClass());
c += updateEntityDB(true, conn, info, act.entity); c += updateEntityDBStatement(true, conn, info, act.entity);
} else if (action instanceof UpdateBatchAction2) { } else if (action instanceof UpdateBatchAction2) {
UpdateBatchAction2 act = (UpdateBatchAction2) action; UpdateBatchAction2 act = (UpdateBatchAction2) action;
EntityInfo info = apply(act.clazz); EntityInfo info = apply(act.clazz);
UpdateSqlInfo sql = updateColumnSql(info, act.pk, act.values); UpdateSqlInfo sql = updateColumnSql(info, act.pk, act.values);
c += updateColumnDB(true, conn, info, null, sql); c += updateColumnDBStatement(true, conn, info, null, sql);
} else if (action instanceof UpdateBatchAction3) { } else if (action instanceof UpdateBatchAction3) {
UpdateBatchAction3 act = (UpdateBatchAction3) action; UpdateBatchAction3 act = (UpdateBatchAction3) action;
EntityInfo info = apply(act.clazz); EntityInfo info = apply(act.clazz);
UpdateSqlInfo sql = updateColumnSql(info, act.node, act.flipper, act.values); UpdateSqlInfo sql = updateColumnSql(info, act.node, act.flipper, act.values);
c += updateColumnDB(true, conn, info, act.flipper, sql); c += updateColumnDBStatement(true, conn, info, act.flipper, sql);
} else if (action instanceof UpdateBatchAction4) { } else if (action instanceof UpdateBatchAction4) {
UpdateBatchAction4 act = (UpdateBatchAction4) action; UpdateBatchAction4 act = (UpdateBatchAction4) action;
EntityInfo info = apply(act.entity.getClass()); EntityInfo info = apply(act.entity.getClass());
UpdateSqlInfo sql = updateColumnSql(info, false, act.entity, act.node, act.selects); UpdateSqlInfo sql = updateColumnSql(info, false, act.entity, act.node, act.selects);
c += updateColumnDB(true, conn, info, null, sql); c += updateColumnDBStatement(true, conn, info, null, sql);
} }
} }
conn.commit(); conn.commit();
@@ -313,7 +318,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
try { try {
conn = writePool.pollConnection(); conn = writePool.pollConnection();
conn.setAutoCommit(false); conn.setAutoCommit(false);
int c = insertDB(false, conn, info, entitys); int c = insertDBStatement(false, conn, info, entitys);
conn.commit(); conn.commit();
return c; return c;
} catch (SQLException e) { } catch (SQLException e) {
@@ -331,7 +336,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
} }
private <T> int insertDB(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, T... entitys) throws SQLException { private <T> int insertDBStatement(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, T... entitys) throws SQLException {
final long s = System.currentTimeMillis(); final long s = System.currentTimeMillis();
int c = 0; int c = 0;
String presql = null; String presql = null;
@@ -347,21 +352,13 @@ public class DataJdbcSource extends AbstractDataSqlSource {
prestmts = prepareInsertEntityStatements(conn, info, prepareInfos, entitys); prestmts = prepareInsertEntityStatements(conn, info, prepareInfos, entitys);
} }
try { try {
if (info.getTableStrategy() == null) { //单库单表 if (info.getTableStrategy() == null) { //单库单表
int c1 = 0; c = Utility.sum(prestmt.executeBatch());
int[] cs = prestmt.executeBatch();
for (int cc : cs) {
c1 += cc;
}
c = c1;
conn.offerUpdateStatement(prestmt); conn.offerUpdateStatement(prestmt);
} else { //分库分表 } else { //分库分表
int c1 = 0; int c1 = 0;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
int[] cs = stmt.executeBatch(); c1 += Utility.sum(stmt.executeBatch());
for (int cc : cs) {
c1 += cc;
}
} }
c = c1; c = c1;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
@@ -494,12 +491,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
if (info.getTableStrategy() == null) { //单库单表 if (info.getTableStrategy() == null) { //单库单表
conn.offerUpdateStatement(prestmt); conn.offerUpdateStatement(prestmt);
prestmt = prepareInsertEntityStatement(conn, presql, info, entitys); prestmt = prepareInsertEntityStatement(conn, presql, info, entitys);
int c1 = 0; c = Utility.sum(prestmt.executeBatch());;
int[] cs = prestmt.executeBatch();
for (int cc : cs) {
c1 += cc;
}
c = c1;
conn.offerUpdateStatement(prestmt); conn.offerUpdateStatement(prestmt);
} else { //分库分表 } else { //分库分表
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
@@ -508,10 +500,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
prestmts = prepareInsertEntityStatements(conn, info, prepareInfos, entitys); prestmts = prepareInsertEntityStatements(conn, info, prepareInfos, entitys);
int c1 = 0; int c1 = 0;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
int[] cs = stmt.executeBatch(); c1 += Utility.sum(stmt.executeBatch());
for (int cc : cs) {
c1 += cc;
}
} }
c = c1; c = c1;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
@@ -594,7 +583,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
try { try {
conn = writePool.pollConnection(); conn = writePool.pollConnection();
conn.setAutoCommit(false); conn.setAutoCommit(false);
int c = deleteDB(false, conn, info, tables, flipper, node, pkmap, sqls); int c = deleteDBStatement(false, conn, info, tables, flipper, node, pkmap, sqls);
conn.commit(); conn.commit();
return c; return c;
} catch (SQLException e) { } catch (SQLException e) {
@@ -612,27 +601,21 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
} }
private <T> int deleteDB(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, String... sqls) throws SQLException { private <T> int deleteDBStatement(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, String... sqls) throws SQLException {
final long s = System.currentTimeMillis(); final long s = System.currentTimeMillis();
try { try {
int c; int c;
if (sqls.length == 1) { if (sqls.length == 1) {
final Statement stmt = conn.createUpdateStatement(); final Statement stmt = conn.createUpdateStatement();
int c1 = stmt.executeUpdate(sqls[0]); c = stmt.executeUpdate(sqls[0]);
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
c = c1;
} else { } else {
final Statement stmt = conn.createUpdateStatement(); final Statement stmt = conn.createUpdateStatement();
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c1 = 0; c = Utility.sum(stmt.executeBatch());
int[] cs = stmt.executeBatch();
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
for (int cc : cs) {
c1 += cc;
}
c = c1;
} }
if (!batch) { if (!batch) {
conn.commit(); conn.commit();
@@ -700,12 +683,8 @@ public class DataJdbcSource extends AbstractDataSqlSource {
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c = 0; int c = Utility.sum(stmt.executeBatch());
int[] cs = stmt.executeBatch();
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
for (int cc : cs) {
c += cc;
}
conn.commit(); conn.commit();
slowLog(s, sqls); slowLog(s, sqls);
return c; return c;
@@ -735,21 +714,15 @@ public class DataJdbcSource extends AbstractDataSqlSource {
int c; int c;
if (sqls.length == 1) { if (sqls.length == 1) {
final Statement stmt = conn.createUpdateStatement(); final Statement stmt = conn.createUpdateStatement();
int c1 = stmt.executeUpdate(sqls[0]); c = stmt.executeUpdate(sqls[0]);
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
c = c1;
} else { } else {
final Statement stmt = conn.createUpdateStatement(); final Statement stmt = conn.createUpdateStatement();
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c1 = 0; c = Utility.sum(stmt.executeBatch());
int[] cs = stmt.executeBatch();
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
for (int cc : cs) {
c1 += cc;
}
c = c1;
} }
conn.commit(); conn.commit();
slowLog(s, sqls); slowLog(s, sqls);
@@ -795,12 +768,8 @@ public class DataJdbcSource extends AbstractDataSqlSource {
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c = 0; int c = Utility.sum(stmt.executeBatch());
int[] cs = stmt.executeBatch();
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
for (int cc : cs) {
c += cc;
}
conn.commit(); conn.commit();
slowLog(s, sqls); slowLog(s, sqls);
return c; return c;
@@ -841,21 +810,15 @@ public class DataJdbcSource extends AbstractDataSqlSource {
if (copyTableSql == null) { if (copyTableSql == null) {
if (sqls.length == 1) { if (sqls.length == 1) {
stmt = conn.createUpdateStatement(); stmt = conn.createUpdateStatement();
int c1 = stmt.executeUpdate(sqls[0]); c = stmt.executeUpdate(sqls[0]);
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
c = c1;
} else { } else {
stmt = conn.createUpdateStatement(); stmt = conn.createUpdateStatement();
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c1 = 0; c = Utility.sum(stmt.executeBatch());
int[] cs = stmt.executeBatch();
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
for (int cc : cs) {
c1 += cc;
}
c = c1;
} }
} else { //建分表 } else { //建分表
try { try {
@@ -968,21 +931,15 @@ public class DataJdbcSource extends AbstractDataSqlSource {
int c; int c;
if (sqls.length == 1) { if (sqls.length == 1) {
final Statement stmt = conn.createUpdateStatement(); final Statement stmt = conn.createUpdateStatement();
int c1 = stmt.executeUpdate(sqls[0]); c = stmt.executeUpdate(sqls[0]);
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
c = c1;
} else { } else {
final Statement stmt = conn.createUpdateStatement(); final Statement stmt = conn.createUpdateStatement();
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c1 = 0; c = Utility.sum(stmt.executeBatch());
int[] cs = stmt.executeBatch();
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
for (int cc : cs) {
c1 += cc;
}
c = c1;
} }
conn.commit(); conn.commit();
slowLog(s, sqls); slowLog(s, sqls);
@@ -1028,12 +985,8 @@ public class DataJdbcSource extends AbstractDataSqlSource {
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c = 0; int c = Utility.sum(stmt.executeBatch());
int[] cs = stmt.executeBatch();
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
for (int cc : cs) {
c += cc;
}
conn.commit(); conn.commit();
slowLog(s, sqls); slowLog(s, sqls);
return c; return c;
@@ -1063,7 +1016,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
try { try {
conn = writePool.pollConnection(); conn = writePool.pollConnection();
conn.setAutoCommit(false); conn.setAutoCommit(false);
int c = updateEntityDB(false, conn, info, entitys); int c = updateEntityDBStatement(false, conn, info, entitys);
conn.commit(); conn.commit();
return c; return c;
} catch (SQLException e) { } catch (SQLException e) {
@@ -1081,7 +1034,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
} }
private <T> int updateEntityDB(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, T... entitys) throws SQLException { private <T> int updateEntityDBStatement(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, T... entitys) throws SQLException {
final long s = System.currentTimeMillis(); final long s = System.currentTimeMillis();
String presql = null; String presql = null;
String caseSql = null; String caseSql = null;
@@ -1188,10 +1141,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
prestmts = prepareUpdateEntityStatements(conn, info, prepareInfos, entitys); prestmts = prepareUpdateEntityStatements(conn, info, prepareInfos, entitys);
int c1 = 0; int c1 = 0;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
int[] cs = stmt.executeBatch(); c1 += Utility.sum(stmt.executeBatch());
for (int cc : cs) {
c1 += cc;
}
} }
c = c1; c = c1;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
@@ -1279,7 +1229,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
try { try {
conn = writePool.pollConnection(); conn = writePool.pollConnection();
conn.setAutoCommit(false); conn.setAutoCommit(false);
int c = updateColumnDB(false, conn, info, flipper, sql); int c = updateColumnDBStatement(false, conn, info, flipper, sql);
conn.commit(); conn.commit();
return c; return c;
} catch (SQLException e) { } catch (SQLException e) {
@@ -1297,7 +1247,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
} }
private <T> int updateColumnDB(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, Flipper flipper, UpdateSqlInfo sql) throws SQLException { //String sql, boolean prepared, Object... blobs) { private <T> int updateColumnDBStatement(final boolean batch, final SourceConnection conn, final EntityInfo<T> info, Flipper flipper, UpdateSqlInfo sql) throws SQLException { //String sql, boolean prepared, Object... blobs) {
final long s = System.currentTimeMillis(); final long s = System.currentTimeMillis();
int c = -1; int c = -1;
String firstTable = null; String firstTable = null;
@@ -1344,10 +1294,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
int c1 = 0; int c1 = 0;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
int[] cs = stmt.executeBatch(); c1 += Utility.sum(stmt.executeBatch());
for (int cc : cs) {
c1 += cc;
}
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
} }
c = c1; c = c1;
@@ -1437,10 +1384,7 @@ public class DataJdbcSource extends AbstractDataSqlSource {
} }
int c1 = 0; int c1 = 0;
for (PreparedStatement stmt : prestmts) { for (PreparedStatement stmt : prestmts) {
int[] cs = stmt.executeBatch(); c1 += Utility.sum(stmt.executeBatch());
for (int cc : cs) {
c1 += cc;
}
conn.offerUpdateStatement(stmt); conn.offerUpdateStatement(stmt);
} }
c = c1; c = c1;