diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index c86b43f0..a1dcfeba 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -23,11 +23,11 @@ canal.conf: # url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true # username: root # password: 121212 -# canalInstances: -# - instance: example -# groups: -# - outAdapters: -# - name: logger + canalInstances: + - instance: example + groups: + - outAdapters: + - name: logger # - name: rdb # key: oracle1 # properties: @@ -35,6 +35,7 @@ canal.conf: # jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc.username: mytest # jdbc.password: m121212 +# threads: 5 # - name: rdb # key: postgres1 # properties: @@ -42,6 +43,7 @@ canal.conf: # jdbc.url: jdbc:postgresql://localhost:5432/postgres # jdbc.username: postgres # jdbc.password: 121212 +# threads: 5 # - name: hbase # properties: # hbase.zookeeper.quorum: 127.0.0.1 diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index 85f2f62c..92bbeebc 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -59,7 +59,7 @@ public class RdbAdapter implements OuterAdapter { dataSource.setPassword(properties.get("jdbc.password")); dataSource.setInitialSize(1); dataSource.setMinIdle(1); - dataSource.setMaxActive(2); + dataSource.setMaxActive(20); dataSource.setMaxWait(60000); dataSource.setTimeBetweenEvictionRunsMillis(60000); dataSource.setMinEvictableIdleTimeMillis(300000); @@ -70,7 +70,12 @@ public class RdbAdapter implements OuterAdapter { logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e); } - rdbSyncService = new RdbSyncService(dataSource); + String threads = properties.get("threads"); + if (threads != null) { + rdbSyncService = new RdbSyncService(Integer.valueOf(threads), dataSource); + } else { + rdbSyncService = new RdbSyncService(null, dataSource); + } } @Override @@ -177,6 +182,9 @@ public class RdbAdapter implements OuterAdapter { @Override public void destroy() { + if (rdbSyncService != null) { + rdbSyncService.close(); + } if (dataSource != null) { dataSource.close(); } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java new file mode 100644 index 00000000..e079c335 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java @@ -0,0 +1,124 @@ +package com.alibaba.otter.canal.client.adapter.rdb.service; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class BatchExecutor { + + private static final Logger logger = LoggerFactory.getLogger(BatchExecutor.class); + + private Integer key; + private Connection conn; + private int commitSize = 3000; + private AtomicInteger idx = new AtomicInteger(0); + private ExecutorService executor = Executors.newFixedThreadPool(1); + private Lock commitLock = new ReentrantLock(); + private Condition condition = commitLock.newCondition(); + + public BatchExecutor(Integer key, Connection conn, Integer commitSize){ + this.key = key; + this.conn = conn; + if (commitSize != null) { + this.commitSize = commitSize; + } + + try { + this.conn.setAutoCommit(false); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + + public Integer getKey() { + return key; + } + + public Connection getConn() { + return conn; + } + + public static void setValue(List> values, int type, Object value) { + Map valueItem = new HashMap<>(); + valueItem.put("type", type); + valueItem.put("value", value); + values.add(valueItem); + } + + public void execute(String sql, List> values) { + try (PreparedStatement pstmt = conn.prepareStatement(sql)) { + int len = values.size(); + for (int i = 0; i < len; i++) { + int type = (Integer) values.get(i).get("type"); + Object value = values.get(i).get("value"); + RdbSyncService.setPStmt(type, pstmt, value, i + 1); + } + + pstmt.execute(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + int i = idx.incrementAndGet(); + + // 批次的第一次执行设置延时 + if (i == 1) { + executor.submit(() -> { + try { + commitLock.lock(); + if (!condition.await(5, TimeUnit.SECONDS)) { + // 超时提交 + commit(); + } + } catch (InterruptedException e) { + // ignore + } finally { + commitLock.unlock(); + } + }); + } + + if (i == commitSize) { + commit(); + } + } + + private void commit() { + try { + commitLock.lock(); + conn.commit(); + if (logger.isTraceEnabled()) { + logger.trace("Batch executor: " + key + " commit " + idx.get() + " rows"); + } + condition.signal(); + idx.set(0); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } finally { + commitLock.unlock(); + } + } + + public void close() { + if (conn != null) { + try { + conn.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + executor.shutdown(); + } +} diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index def43637..b4a6ecba 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -9,7 +9,7 @@ import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.*; import javax.sql.DataSource; @@ -37,10 +37,30 @@ public class RdbSyncService { private final Map> COLUMNS_TYPE_CACHE = new ConcurrentHashMap<>(); - private DataSource dataSource; + private BatchExecutor[] batchExecutors; - public RdbSyncService(DataSource dataSource){ - this.dataSource = dataSource; + private int threads = 1; + + private ExecutorService[] threadExecutors; + + public RdbSyncService(Integer threads, DataSource dataSource){ + try { + if (threads != null && threads > 1 && threads < 10) { + this.threads = threads; + } + batchExecutors = new BatchExecutor[this.threads]; + for (int i = 0; i < this.threads; i++) { + Connection conn = dataSource.getConnection(); + conn.setAutoCommit(false); + this.batchExecutors[i] = new BatchExecutor(i, conn, null); + } + threadExecutors = new ExecutorService[this.threads]; + for (int i = 0; i < this.threads; i++) { + threadExecutors[i] = Executors.newFixedThreadPool(1); + } + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } } public void sync(MappingConfig config, Dml dml) { @@ -69,7 +89,7 @@ public class RdbSyncService { * @param config 配置项 * @param dml DML数据 */ - private void insert(MappingConfig config, Dml dml) throws SQLException { + private void insert(MappingConfig config, Dml dml) { List> data = dml.getData(); if (data == null || data.isEmpty()) { return; @@ -77,75 +97,52 @@ public class RdbSyncService { DbMapping dbMapping = config.getDbMapping(); - int idx = 1; - boolean completed = false; + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); - Connection conn = dataSource.getConnection(); - conn.setAutoCommit(false); - try { - Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); + StringBuilder insertSql = new StringBuilder(); + insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); - StringBuilder insertSql = new StringBuilder(); - insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); + columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); + int len = insertSql.length(); + insertSql.delete(len - 1, len).append(") VALUES ("); + int mapLen = columnsMap.size(); + for (int i = 0; i < mapLen; i++) { + insertSql.append("?,"); + } + len = insertSql.length(); + insertSql.delete(len - 1, len).append(")"); - columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); - int len = insertSql.length(); - insertSql.delete(len - 1, len).append(") VALUES ("); - int mapLen = columnsMap.size(); - for (int i = 0; i < mapLen; i++) { - insertSql.append("?,"); - } - len = insertSql.length(); - insertSql.delete(len - 1, len).append(")"); + Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); - Map ctype = getTargetColumnType(conn, config); + String sql = insertSql.toString(); + for (Map d : data) { + int pkHash = pkHash(dbMapping, d, threads); + ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[pkHash]; + checkQueue(tpe); + tpe.submit(() -> { + try { + BatchExecutor batchExecutor = batchExecutors[pkHash]; + List> values = new ArrayList<>(); - PreparedStatement pstmt = conn.prepareStatement(insertSql.toString()); - - for (Map d : data) { - pstmt.clearParameters(); - int i = 1; - for (Map.Entry entry : columnsMap.entrySet()) { - String targetClolumnName = entry.getKey(); - String srcColumnName = entry.getValue(); - if (srcColumnName == null) { - srcColumnName = targetClolumnName; - } - - Integer type = ctype.get(targetClolumnName.toLowerCase()); - - Object value = d.get(srcColumnName); - if (value != null) { - if (type == null) { - throw new RuntimeException("No column: " + targetClolumnName + " found in target db"); + for (Map.Entry entry : columnsMap.entrySet()) { + String targetColumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetColumnName; } - setPStmt(type, pstmt, value, i); - } else { - pstmt.setNull(i, type); + Integer type = ctype.get(targetColumnName.toLowerCase()); + if (type == null) { + throw new RuntimeException("No column: " + targetColumnName + " found in target db"); + } + Object value = d.get(srcColumnName); + BatchExecutor.setValue(values, type, value); } - i++; + batchExecutor.execute(sql, values); + } catch (Exception e) { + logger.error(e.getMessage(), e); } - - pstmt.execute(); - if (logger.isTraceEnabled()) { - logger.trace("Insert into target table, sql: {}", insertSql); - } - - if (idx % config.getDbMapping().getCommitBatch() == 0) { - conn.commit(); - completed = true; - } - idx++; - } - if (!completed) { - conn.commit(); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - conn.rollback(); - } finally { - conn.close(); + }); } } @@ -155,7 +152,7 @@ public class RdbSyncService { * @param config 配置项 * @param dml DML数据 */ - private void update(MappingConfig config, Dml dml) throws SQLException { + private void update(MappingConfig config, Dml dml) { List> data = dml.getData(); if (data == null || data.isEmpty()) { return; @@ -169,61 +166,55 @@ public class RdbSyncService { DbMapping dbMapping = config.getDbMapping(); int idx = 1; - boolean completed = false; + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); - Connection conn = dataSource.getConnection(); - conn.setAutoCommit(false); + Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); - try { - Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); + for (Map o : old) { + Map d = data.get(idx - 1); + int pkHash = pkHash(dbMapping, d, threads); - Map ctype = getTargetColumnType(conn, config); + ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[pkHash]; + checkQueue(tpe); + tpe.submit(() -> { + try { + BatchExecutor batchExecutor = batchExecutors[pkHash]; - for (Map o : old) { - Map d = data.get(idx - 1); - StringBuilder updateSql = new StringBuilder(); - updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); - Map values = new LinkedHashMap<>(); - for (String srcColumnName : o.keySet()) { - List targetColumnNames = new ArrayList<>(); - columnsMap.forEach((targetColumn, srcColumn) -> { - if (srcColumnName.toLowerCase().equals(srcColumn)) { - targetColumnNames.add(targetColumn); - } - }); - if (!targetColumnNames.isEmpty()) { + StringBuilder updateSql = new StringBuilder(); + updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); - for (String targetColumnName : targetColumnNames) { - updateSql.append(targetColumnName).append("=?, "); - values.put(targetColumnName, d.get(srcColumnName)); + List> values = new ArrayList<>(); + for (String srcColumnName : o.keySet()) { + List targetColumnNames = new ArrayList<>(); + columnsMap.forEach((targetColumn, srcColumn) -> { + if (srcColumnName.toLowerCase().equals(srcColumn)) { + targetColumnNames.add(targetColumn); + } + }); + if (!targetColumnNames.isEmpty()) { + for (String targetColumnName : targetColumnNames) { + updateSql.append(targetColumnName).append("=?, "); + Integer type = ctype.get(targetColumnName.toLowerCase()); + BatchExecutor.setValue(values, type, d.get(srcColumnName)); + } } } - } - int len = updateSql.length(); - updateSql.delete(len - 2, len).append(" WHERE "); + int len = updateSql.length(); + updateSql.delete(len - 2, len).append(" WHERE "); - // 拼接主键 - appendCondition(dbMapping, updateSql, values, d); + // 拼接主键 + appendCondition(dbMapping, updateSql, ctype, values, d); - sqlExe(conn, updateSql.toString(), ctype, values); + if (logger.isTraceEnabled()) { + logger.trace("Update target table, sql: {}", updateSql); + } + batchExecutor.execute(updateSql.toString(), values); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + }); - if (logger.isTraceEnabled()) { - logger.trace("Update target table, sql: {}", updateSql); - } - if (idx % config.getDbMapping().getCommitBatch() == 0) { - conn.commit(); - completed = true; - } - idx++; - } - if (!completed) { - conn.commit(); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - conn.rollback(); - } finally { - conn.close(); + idx++; } } @@ -234,7 +225,7 @@ public class RdbSyncService { * @param dml * @throws SQLException */ - private void delete(MappingConfig config, Dml dml) throws SQLException { + private void delete(MappingConfig config, Dml dml) { List> data = dml.getData(); if (data == null || data.isEmpty()) { return; @@ -242,42 +233,32 @@ public class RdbSyncService { DbMapping dbMapping = config.getDbMapping(); - int idx = 1; - boolean completed = false; + Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); - Connection conn = dataSource.getConnection(); - conn.setAutoCommit(false); + for (Map d : data) { + int pkHash = pkHash(dbMapping, d, threads); - try { - Map ctype = getTargetColumnType(conn, config); + ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[pkHash]; + checkQueue(tpe); + tpe.submit(() -> { + try { + BatchExecutor batchExecutor = batchExecutors[pkHash]; + StringBuilder sql = new StringBuilder(); + sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); - for (Map d : data) { - StringBuilder sql = new StringBuilder(); - sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); + List> values = new ArrayList<>(); - Map values = new LinkedHashMap<>(); - // 拼接主键 - appendCondition(dbMapping, sql, values, d); + // 拼接主键 + appendCondition(dbMapping, sql, ctype, values, d); - sqlExe(conn, sql.toString(), ctype, values); - - if (logger.isTraceEnabled()) { - logger.trace("Delete from target table, sql: {}", sql); + if (logger.isTraceEnabled()) { + logger.trace("Delete from target table, sql: {}", sql); + } + batchExecutor.execute(sql.toString(), values); + } catch (Exception e) { + logger.error(e.getMessage(), e); } - if (idx % config.getDbMapping().getCommitBatch() == 0) { - conn.commit(); - completed = true; - } - idx++; - } - if (!completed) { - conn.commit(); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - conn.rollback(); - } finally { - conn.close(); + }); } } @@ -326,6 +307,9 @@ public class RdbSyncService { * @param i 索引号 */ public static void setPStmt(int type, PreparedStatement pstmt, Object value, int i) throws SQLException { + if (value == null) { + pstmt.setObject(i, type); + } switch (type) { case Types.BIT: case Types.BOOLEAN: @@ -431,7 +415,7 @@ public class RdbSyncService { break; case Types.DATE: if (value instanceof java.util.Date) { - pstmt.setDate(i, new java.sql.Date(((java.util.Date) value).getTime())); + pstmt.setDate(i, new Date(((java.util.Date) value).getTime())); } else if (value instanceof String) { String v = (String) value; if (!v.startsWith("0000-00-00")) { @@ -447,7 +431,7 @@ public class RdbSyncService { break; case Types.TIME: if (value instanceof java.util.Date) { - pstmt.setTime(i, new java.sql.Time(((java.util.Date) value).getTime())); + pstmt.setTime(i, new Time(((java.util.Date) value).getTime())); } else if (value instanceof String) { String v = (String) value; v = "T" + v; @@ -459,7 +443,7 @@ public class RdbSyncService { break; case Types.TIMESTAMP: if (value instanceof java.util.Date) { - pstmt.setTimestamp(i, new java.sql.Timestamp(((java.util.Date) value).getTime())); + pstmt.setTimestamp(i, new Timestamp(((java.util.Date) value).getTime())); } else if (value instanceof String) { String v = (String) value; if (!v.startsWith("0000-00-00")) { @@ -481,8 +465,8 @@ public class RdbSyncService { /** * 拼接主键 where条件 */ - private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map values, - Map d) { + private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map ctype, + List> values, Map d) { // 拼接主键 for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) { String targetColumnName = entry.getKey(); @@ -491,28 +475,54 @@ public class RdbSyncService { srcColumnName = targetColumnName; } sql.append(targetColumnName).append("=? AND "); - values.put(targetColumnName, d.get(srcColumnName)); + Integer type = ctype.get(targetColumnName.toLowerCase()); + BatchExecutor.setValue(values, type, d.get(srcColumnName)); } int len = sql.length(); sql.delete(len - 4, len); } /** - * 执行sql + * 取主键hash */ - private static void sqlExe(Connection conn, String sql, Map ctype, Map values) { - try (PreparedStatement pstmt = conn.prepareStatement(sql)) { - int i = 1; - for (Map.Entry entry : values.entrySet()) { - String targetColumnName = entry.getKey(); - Object value = entry.getValue(); - Integer type = ctype.get(targetColumnName.toLowerCase()); - setPStmt(type, pstmt, value, i++); + private static int pkHash(DbMapping dbMapping, Map d, int threads) { + int hash = 0; + // 取主键 + for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) { + String targetColumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetColumnName; } - pstmt.execute(); - } catch (SQLException e) { - logger.error(e.getMessage(), e); + Object value = d.get(srcColumnName); + if (value != null) { + hash += value.hashCode(); + } + } + hash = Math.abs(hash) % threads; + return Math.abs(hash); + } + + public void close() { + for (BatchExecutor batchExecutor : batchExecutors) { + batchExecutor.close(); + } + for (ExecutorService executorService : threadExecutors) { + executorService.shutdown(); } } + private void checkQueue(ThreadPoolExecutor tpe) { + // 防止队列过大 + while (tpe.getQueue().size() > 10000) { + try { + Thread.sleep(3000); + while (tpe.getQueue().size() > 5000) { + Thread.sleep(1000); + } + } catch (InterruptedException e) { + // ignore + } + } + } } diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java index 54e6c447..b4e4a02e 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java @@ -2,8 +2,10 @@ package com.alibaba.otter.canal.client.adapter.rdb.test; import java.io.BufferedReader; import java.io.Reader; -import java.math.BigDecimal; -import java.sql.*; +import java.sql.Clob; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; import org.junit.Test; @@ -34,30 +36,55 @@ public class DBTest { dataSource.init(); Connection conn = dataSource.getConnection(); - Statement stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery("select * from user t where 1=2"); - ResultSetMetaData rsm = rs.getMetaData(); - int cnt = rsm.getColumnCount(); - for (int i = 1; i <= cnt; i++) { - System.out.println(rsm.getColumnName(i) + " " + rsm.getColumnType(i)); + conn.setAutoCommit(false); + PreparedStatement pstmt = conn + .prepareStatement("insert into user (id,name,role_id,c_time,test1,test2) values (?,?,?,?,?,?)"); + + java.util.Date now = new java.util.Date(); + for (int i = 1; i <= 100000; i++) { + pstmt.clearParameters(); + pstmt.setLong(1, (long) i); + pstmt.setString(2, "test_" + i); + pstmt.setLong(3, (long) i % 4 + 1); + pstmt.setDate(4, new java.sql.Date(now.getTime())); + pstmt.setString(5, "tttt"); + pstmt.setBytes(6, null); + + pstmt.execute(); + if (i % 5000 == 0) { + conn.commit(); + } } + conn.commit(); - rs.close(); - stmt.close(); + pstmt.close(); -// PreparedStatement pstmt = conn -// .prepareStatement("insert into tb_user (id,name,role_id,c_time,test1,test2) values (?,?,?,?,?,?)"); -// pstmt.setBigDecimal(1, new BigDecimal("5")); -// pstmt.setString(2, "test"); -// pstmt.setBigDecimal(3, new BigDecimal("1")); -// pstmt.setDate(4, new Date(new java.util.Date().getTime())); -// byte[] a = { (byte) 1, (byte) 2 }; -// pstmt.setBytes(5, a); -// pstmt.setBytes(6, a); -// pstmt.execute(); -// -// pstmt.close(); + // Statement stmt = conn.createStatement(); + // ResultSet rs = stmt.executeQuery("select * from user t where 1=2"); + // + // ResultSetMetaData rsm = rs.getMetaData(); + // int cnt = rsm.getColumnCount(); + // for (int i = 1; i <= cnt; i++) { + // System.out.println(rsm.getColumnName(i) + " " + rsm.getColumnType(i)); + // } + + // rs.close(); + // stmt.close(); + + // PreparedStatement pstmt = conn + // .prepareStatement("insert into tb_user (id,name,role_id,c_time,test1,test2) + // values (?,?,?,?,?,?)"); + // pstmt.setBigDecimal(1, new BigDecimal("5")); + // pstmt.setString(2, "test"); + // pstmt.setBigDecimal(3, new BigDecimal("1")); + // pstmt.setDate(4, new Date(new java.util.Date().getTime())); + // byte[] a = { (byte) 1, (byte) 2 }; + // pstmt.setBytes(5, a); + // pstmt.setBytes(6, a); + // pstmt.execute(); + // + // pstmt.close(); conn.close(); dataSource.close();